






import json
import os
from pathlib import Path
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple, Union

import sentencepiece

from ...tokenization_utils import BatchEncoding, PreTrainedTokenizer
from ...utils import logging


_a    :   Any        =			logging.get_logger(__name__)

_a    :   Union[str, Any]        =			"""▁"""

_a    :   Dict        =			{
    """vocab_file""": """vocab.json""",
    """spm_file""": """sentencepiece.bpe.model""",
    """tokenizer_config_file""": """tokenizer_config.json""",
}

_a    :   Optional[int]        =			{
    """vocab_file""": {
        """facebook/m2m100_418M""": """https://huggingface.co/facebook/m2m100_418M/resolve/main/vocab.json""",
        """facebook/m2m100_1.2B""": """https://huggingface.co/facebook/m2m100_1.2B/resolve/main/vocab.json""",
    },
    """spm_file""": {
        """facebook/m2m100_418M""": """https://huggingface.co/facebook/m2m100_418M/resolve/main/sentencepiece.bpe.model""",
        """facebook/m2m100_1.2B""": """https://huggingface.co/facebook/m2m100_1.2B/resolve/main/sentencepiece.bpe.model""",
    },
    """tokenizer_config_file""": {
        """facebook/m2m100_418M""": """https://huggingface.co/facebook/m2m100_418M/resolve/main/tokenizer_config.json""",
        """facebook/m2m100_1.2B""": """https://huggingface.co/facebook/m2m100_1.2B/resolve/main/tokenizer_config.json""",
    },
}

_a    :   List[Any]        =			{
    """facebook/m2m100_418M""": 1024,
}

# fmt: off
_a    :   List[str]        =			{
    """m2m100""": ["""af""", """am""", """ar""", """ast""", """az""", """ba""", """be""", """bg""", """bn""", """br""", """bs""", """ca""", """ceb""", """cs""", """cy""", """da""", """de""", """el""", """en""", """es""", """et""", """fa""", """ff""", """fi""", """fr""", """fy""", """ga""", """gd""", """gl""", """gu""", """ha""", """he""", """hi""", """hr""", """ht""", """hu""", """hy""", """id""", """ig""", """ilo""", """is""", """it""", """ja""", """jv""", """ka""", """kk""", """km""", """kn""", """ko""", """lb""", """lg""", """ln""", """lo""", """lt""", """lv""", """mg""", """mk""", """ml""", """mn""", """mr""", """ms""", """my""", """ne""", """nl""", """no""", """ns""", """oc""", """or""", """pa""", """pl""", """ps""", """pt""", """ro""", """ru""", """sd""", """si""", """sk""", """sl""", """so""", """sq""", """sr""", """ss""", """su""", """sv""", """sw""", """ta""", """th""", """tl""", """tn""", """tr""", """uk""", """ur""", """uz""", """vi""", """wo""", """xh""", """yi""", """yo""", """zh""", """zu"""],
    """wmt21""": ["""en""", """ha""", """is""", """ja""", """cs""", """ru""", """zh""", """de"""]
}







class 	_UpperCAmelCase  (  _A		):




      """simple docstring"""



      A          =    VOCAB_FILES_NAMES
      A          =    PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      A          =    PRETRAINED_VOCAB_FILES_MAP
      A          =    ['''input_ids''', '''attention_mask''']

      A          =    []
      A          =    []
      def __init__(   self    ,	_lowerCAmelCase    ,	_lowerCAmelCase    ,	_lowerCAmelCase=None    ,	_lowerCAmelCase=None    ,	_lowerCAmelCase="<s>"    ,	_lowerCAmelCase="</s>"    ,	_lowerCAmelCase="</s>"    ,	_lowerCAmelCase="<pad>"    ,	_lowerCAmelCase="<unk>"    ,	_lowerCAmelCase="m2m100"    ,	_lowerCAmelCase = None    ,	_lowerCAmelCase=8    ,	**_lowerCAmelCase    ,	):



          '''simple docstring'''
          lowerCAmelCase__   :int      							=				{} if sp_model_kwargs is None else sp_model_kwargs

          lowerCAmelCase__   :int      							=				language_codes
          lowerCAmelCase__   :Optional[Any]      							=				FAIRSEQ_LANGUAGE_CODES[language_codes]
          lowerCAmelCase__   :Dict      							=				{lang_code: f'''__{lang_code}__''' for lang_code in fairseq_language_code}

          lowerCAmelCase__   :Optional[int]      							=				kwargs.get("additional_special_tokens"    ,	[]		)
          kwargs["additional_special_tokens"] += [
              self.get_lang_token(_lowerCAmelCase		)
              for lang_code in fairseq_language_code
              if self.get_lang_token(_lowerCAmelCase		) not in kwargs["additional_special_tokens"]
          ]

          super().__init__(
              src_lang=_lowerCAmelCase    ,	tgt_lang=_lowerCAmelCase    ,	bos_token=_lowerCAmelCase    ,	eos_token=_lowerCAmelCase    ,	sep_token=_lowerCAmelCase    ,	unk_token=_lowerCAmelCase    ,	pad_token=_lowerCAmelCase    ,	language_codes=_lowerCAmelCase    ,	sp_model_kwargs=self.sp_model_kwargs    ,	num_madeup_words=_lowerCAmelCase    ,	**_lowerCAmelCase    ,	)

          lowerCAmelCase__   :Union[str, Any]      							=				vocab_file
          lowerCAmelCase__   :Tuple      							=				load_json(_lowerCAmelCase		)
          lowerCAmelCase__   :List[str]      							=				{v: k for k, v in self.encoder.items()}
          lowerCAmelCase__   :str      							=				spm_file
          lowerCAmelCase__   :Union[str, Any]      							=				load_spm(_lowerCAmelCase    ,	self.sp_model_kwargs		)

          lowerCAmelCase__   :Optional[int]      							=				len(self.encoder		)

          lowerCAmelCase__   :Any      							=				{
              self.get_lang_token(_lowerCAmelCase		): self.encoder_size + i for i, lang_code in enumerate(_lowerCAmelCase		)
          }
          lowerCAmelCase__   :Optional[Any]      							=				{lang_code: self.encoder_size + i for i, lang_code in enumerate(_lowerCAmelCase		)}
          lowerCAmelCase__   :Any      							=				{v: k for k, v in self.lang_token_to_id.items()}

          lowerCAmelCase__   :List[str]      							=				src_lang if src_lang is not None else "en"
          lowerCAmelCase__   :int      							=				tgt_lang
          lowerCAmelCase__   :Optional[int]      							=				self.get_lang_id(self._src_lang		)
          self.set_src_lang_special_tokens(self._src_lang		)

          lowerCAmelCase__   :int      							=				num_madeup_words
      @property
      def 			snake_case_      (   self		):



          '''simple docstring'''
          return len(self.encoder		) + len(self.lang_token_to_id		)
      @property
      def 			snake_case_      (   self		):



          '''simple docstring'''
          return self._src_lang
      @src_lang.setter
      def 			snake_case_      (   self    ,	_lowerCAmelCase		):



          '''simple docstring'''
          lowerCAmelCase__   :List[str]      							=				new_src_lang
          self.set_src_lang_special_tokens(self._src_lang		)
      def 			snake_case_      (   self    ,	_lowerCAmelCase		):



          '''simple docstring'''
          return self.sp_model.encode(_lowerCAmelCase    ,	out_type=_lowerCAmelCase		)
      def 			snake_case_      (   self    ,	_lowerCAmelCase		):



          '''simple docstring'''
          if token in self.lang_token_to_id:
              return self.lang_token_to_id[token]
          return self.encoder.get(_lowerCAmelCase    ,	self.encoder[self.unk_token]		)
      def 			snake_case_      (   self    ,	_lowerCAmelCase		):



          '''simple docstring'''
          if index in self.id_to_lang_token:
              return self.id_to_lang_token[index]
          return self.decoder.get(_lowerCAmelCase    ,	self.unk_token		)
      def 			snake_case_      (   self    ,	_lowerCAmelCase		):



          '''simple docstring'''
          lowerCAmelCase__   :List[Any]      							=				[]
          lowerCAmelCase__   :str      							=				""
          for token in tokens:
              # make sure that special tokens are not decoded using sentencepiece model
              if token in self.all_special_tokens:
                  out_string += self.sp_model.decode(_lowerCAmelCase		) + token
                  lowerCAmelCase__   :Any      							=				[]
              else:
                  current_sub_tokens.append(_lowerCAmelCase		)
          out_string += self.sp_model.decode(_lowerCAmelCase		)
          return out_string.strip()
      def 			snake_case_      (   self    ,	_lowerCAmelCase    ,	_lowerCAmelCase = None    ,	_lowerCAmelCase = False		):



          '''simple docstring'''

          if already_has_special_tokens:
              return super().get_special_tokens_mask(
                  token_ids_a=_lowerCAmelCase    ,	token_ids_a=_lowerCAmelCase    ,	already_has_special_tokens=_lowerCAmelCase		)

          lowerCAmelCase__   :Union[str, Any]      							=				[1] * len(self.prefix_tokens		)
          lowerCAmelCase__   :Union[str, Any]      							=				[1] * len(self.suffix_tokens		)
          if token_ids_a is None:
              return prefix_ones + ([0] * len(_lowerCAmelCase		)) + suffix_ones
          return prefix_ones + ([0] * len(_lowerCAmelCase		)) + ([0] * len(_lowerCAmelCase		)) + suffix_ones
      def 			snake_case_      (   self    ,	_lowerCAmelCase    ,	_lowerCAmelCase = None		):



          '''simple docstring'''
          if token_ids_a is None:
              return self.prefix_tokens + token_ids_a + self.suffix_tokens
          # We don't expect to process pairs, but leave the pair logic for API consistency
          return self.prefix_tokens + token_ids_a + token_ids_a + self.suffix_tokens
      def 			snake_case_      (   self		):



          '''simple docstring'''
          lowerCAmelCase__   :Union[str, Any]      							=				{self.convert_ids_to_tokens(_lowerCAmelCase		): i for i in range(self.vocab_size		)}
          vocab.update(self.added_tokens_encoder		)
          return vocab
      def __getstate__(   self		):



          '''simple docstring'''
          lowerCAmelCase__   :List[str]      							=				self.__dict__.copy()
          lowerCAmelCase__   :int      							=				None
          return state
      def __setstate__(   self    ,	_lowerCAmelCase		):



          '''simple docstring'''
          lowerCAmelCase__   :Dict      							=				d

          # for backward compatibility
          if not hasattr(self    ,	"sp_model_kwargs"		):
              lowerCAmelCase__   :str      							=				{}

          lowerCAmelCase__   :List[Any]      							=				load_spm(self.spm_file    ,	self.sp_model_kwargs		)
      def 			snake_case_      (   self    ,	_lowerCAmelCase    ,	_lowerCAmelCase = None		):



          '''simple docstring'''
          lowerCAmelCase__   :Union[str, Any]      							=				Path(_lowerCAmelCase		)
          if not save_dir.is_dir():
              raise OSError(f'''{save_directory} should be a directory'''		)
          lowerCAmelCase__   :Union[str, Any]      							=				save_dir / (
              (filename_prefix + "-" if filename_prefix else "") + self.vocab_files_names["vocab_file"]
          )
          lowerCAmelCase__   :List[Any]      							=				save_dir / (
              (filename_prefix + "-" if filename_prefix else "") + self.vocab_files_names["spm_file"]
          )

          save_json(self.encoder    ,	_lowerCAmelCase		)

          if os.path.abspath(self.spm_file		) != os.path.abspath(_lowerCAmelCase		) and os.path.isfile(self.spm_file		):
              copyfile(self.spm_file    ,	_lowerCAmelCase		)
          elif not os.path.isfile(self.spm_file		):
              with open(_lowerCAmelCase    ,	"wb"		) as fi:
                  lowerCAmelCase__   :int      							=				self.sp_model.serialized_model_proto()
                  fi.write(_lowerCAmelCase		)

          return (str(_lowerCAmelCase		), str(_lowerCAmelCase		))
      def 			snake_case_      (   self    ,	_lowerCAmelCase    ,	_lowerCAmelCase = "en"    ,	_lowerCAmelCase = None    ,	_lowerCAmelCase = "ro"    ,	**_lowerCAmelCase    ,	):



          '''simple docstring'''
          lowerCAmelCase__   :int      							=				src_lang
          lowerCAmelCase__   :int      							=				tgt_lang
          self.set_src_lang_special_tokens(self.src_lang		)
          return super().prepare_seqaseq_batch(_lowerCAmelCase    ,	_lowerCAmelCase    ,	**_lowerCAmelCase		)
      def 			snake_case_      (   self    ,	_lowerCAmelCase    ,	_lowerCAmelCase    ,	_lowerCAmelCase    ,	**_lowerCAmelCase		):



          '''simple docstring'''
          if src_lang is None or tgt_lang is None:
              raise ValueError("Translation requires a `src_lang` and a `tgt_lang` for this model"		)
          lowerCAmelCase__   :Any      							=				src_lang
          lowerCAmelCase__   :Any      							=				self(_lowerCAmelCase    ,	add_special_tokens=_lowerCAmelCase    ,	**_lowerCAmelCase		)
          lowerCAmelCase__   :int      							=				self.get_lang_id(_lowerCAmelCase		)
          lowerCAmelCase__   :Dict      							=				tgt_lang_id
          return inputs
      def 			snake_case_      (   self		):



          '''simple docstring'''
          self.set_src_lang_special_tokens(self.src_lang		)
      def 			snake_case_      (   self		):



          '''simple docstring'''
          self.set_tgt_lang_special_tokens(self.tgt_lang		)
      def 			snake_case_      (   self    ,	_lowerCAmelCase		):



          '''simple docstring'''
          lowerCAmelCase__   :str      							=				self.get_lang_token(_lowerCAmelCase		)
          lowerCAmelCase__   :List[Any]      							=				self.lang_token_to_id[lang_token]
          lowerCAmelCase__   :str      							=				[self.cur_lang_id]
          lowerCAmelCase__   :Any      							=				[self.eos_token_id]
      def 			snake_case_      (   self    ,	_lowerCAmelCase		):



          '''simple docstring'''
          lowerCAmelCase__   :List[Any]      							=				self.get_lang_token(_lowerCAmelCase		)
          lowerCAmelCase__   :List[Any]      							=				self.lang_token_to_id[lang_token]
          lowerCAmelCase__   :Dict      							=				[self.cur_lang_id]
          lowerCAmelCase__   :List[Any]      							=				[self.eos_token_id]
      def 			snake_case_      (   self    ,	_lowerCAmelCase		):



          '''simple docstring'''
          return self.lang_code_to_token[lang]



      def 			snake_case_      (   self    ,	_lowerCAmelCase		):



          '''simple docstring'''
          lowerCAmelCase__   :Optional[int]      							=				self.get_lang_token(_lowerCAmelCase		)
          return self.lang_token_to_id[lang_token]





def 				snake_case__							(							UpperCAmelCase							:  str    ,			UpperCAmelCase							:  Dict[str, Any]       ):
    lowerCAmelCase__   :List[Any]      							=				sentencepiece.SentencePieceProcessor(**UpperCAmelCase       )
    spm.Load(str(UpperCAmelCase       )       )
    return spm


def 				snake_case__							(							UpperCAmelCase							:  str       ):
    with open(UpperCAmelCase    ,			"r"       ) as f:
        return json.load(UpperCAmelCase       )


def 				snake_case__							(							UpperCAmelCase							:  List[str]    ,			UpperCAmelCase							:  str       ):
    with open(UpperCAmelCase    ,			"w"       ) as f:
        json.dump(UpperCAmelCase    ,			UpperCAmelCase    ,			indent=2       )











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		)



