






'''simple docstring'''
import collections
import json
import os
import re
from typing import TYPE_CHECKING, List, Optional, Tuple

import numpy as np

from ...tokenization_utils_fast import PreTrainedTokenizer
from ...utils import logging


if TYPE_CHECKING:
 from transformers.pipelines.conversational import Conversation


UpperCamelCase__			=      logging.get_logger(__name__)

UpperCamelCase__			=      {'vocab_file': 'vocab.txt', 'emoji_file': 'emoji.json'}

UpperCamelCase__			=      {
    'vocab_file': {
        'abeja/gpt-neox-japanese-2.7b': 'https://huggingface.co/abeja/gpt-neox-japanese-2.7b/resolve/main/vocab.txt',
    },
    'emoji_file': {
        'abeja/gpt-neox-japanese-2.7b': 'https://huggingface.co/abeja/gpt-neox-japanese-2.7b/resolve/main/emoji.json',
    },
}

UpperCamelCase__			=      {
    'abeja/gpt-neox-japanese-2.7b': 2048,
}




def 		__SCREAMING_SNAKE_CASE       (	_UpperCamelCase    ,		_UpperCamelCase				):


   """simple docstring"""




   with open(_UpperCamelCase    ,		"r"    ,		encoding="utf-8"				) as f:
      lowercase_						:     Tuple							     =  json.loads(f.read()				)

   lowercase_						:     int							     =  collections.OrderedDict()
   lowercase_						:     Dict							     =  collections.OrderedDict()
   lowercase_						:     Tuple							     =  collections.OrderedDict()
   with open(_UpperCamelCase    ,		"r"    ,		encoding="utf-8"				) as f:
      lowercase_						:     Optional[Any]							     =  f.readlines()
   lowercase_						:     Tuple							     =  [[t.rstrip("\n"				)] if (t == "," or "," not in t) else t.rstrip("\n"				).split(","				) for t in token]
   for idx, b in enumerate(_UpperCamelCase				):
      lowercase_						:     List[str]							     =  b
      lowercase_						:     Tuple							     =  idx
      for wd in b:
         lowercase_						:     Tuple							     =  idx

   return vocab, raw_vocab, ids_to_tokens, emoji


class 							_UpperCAmelCase    (		snake_case      ):
    __lowerCamelCase: Optional[Any]									=     VOCAB_FILES_NAMES
    __lowerCamelCase: str									=     PRETRAINED_VOCAB_FILES_MAP
    __lowerCamelCase: Union[str, Any]									=     PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
    __lowerCamelCase: str									=     ['input_ids', 'attention_mask']




    def __init__(      self	:							Union[str, Any]	,  a	:							Optional[int]	,  a	:							Tuple	,  a	:							List[str]="<|endoftext|>"	,  a	:							str="<|endoftext|>"	,  a	:							Optional[Any]="<|startoftext|>"	,  a	:							Optional[int]="<|endoftext|>"	,  a	:							Dict=False	,  **a	:							Any	,  ):






       '''simple docstring'''






       super().__init__(
           unk_token=a	,  pad_token=a	,  bos_token=a	,  eos_token=a	,  do_clean_text=a	,  **a	,  )
       if not os.path.isfile(a      ):
          raise ValueError(
              f"""Can't find a vocabulary file at path '{vocab_file}'. To load the vocabulary from a Google pretrained"""
              " model use `tokenizer = GPTNeoXJapaneseokenizer.from_pretrained(PRETRAINED_MODEL_NAME)`"      )
       if not os.path.isfile(a      ):
          raise ValueError(
              f"""Can't find a emoji file at path '{emoji_file}'. To load the emoji information from a Google"""
              " pretrained model use `tokenizer = GPTNeoXJapaneseokenizer.from_pretrained(PRETRAINED_MODEL_NAME)`"      )
       lowercase_						:     Optional[Any]							     =  do_clean_text
       lowercase_     ,		lowercase_     ,		lowercase_     ,		lowercase_						:     Tuple							     =  load_vocab_and_emoji(a	,  a      )
       lowercase_						:     List[str]							     =  SubWordJapaneseTokenizer(
           vocab=self.vocab	,  ids_to_tokens=self.ids_to_tokens	,  emoji=self.emoji      )




    @property
    def 					lowerCAmelCase__					(      self	:							int      ):






       '''simple docstring'''






       return len(self.raw_vocab      )




    def 					lowerCAmelCase__					(      self	:							Any      ):






       '''simple docstring'''






       return dict(self.raw_vocab	,  **self.added_tokens_encoder      )




    def 					lowerCAmelCase__					(      self	:							Optional[int]	,  a	:							Tuple      ):






       '''simple docstring'''






       return self.subword_tokenizer.tokenize(a	,  clean=self.do_clean_text      )




    def 					lowerCAmelCase__					(      self	:							Optional[Any]	,  a	:							Optional[int]      ):






       '''simple docstring'''






       return self.vocab.get(a	,  self.vocab.get(self.unk_token      )      )




    def 					lowerCAmelCase__					(      self	:							Tuple	,  a	:							Optional[int]      ):






       '''simple docstring'''






       return self.subword_tokenizer.convert_id_to_token(a      )




    def 					lowerCAmelCase__					(      self	:							Dict	,  a	:							List[str]      ):






       '''simple docstring'''






       lowercase_						:     Dict							     =  "".join(a      ).strip()
       return out_string




    def 					lowerCAmelCase__					(      self	:							Union[str, Any]	,  a	:							"Conversation"      ):






       '''simple docstring'''






       lowercase_						:     Any							     =  []
       for is_user, text in conversation.iter_texts():
          input_ids.extend(self.encode(a	,  add_special_tokens=a      ) + [self.eos_token_id]      )

       if len(a      ) > self.model_max_length:
          lowercase_						:     Optional[Any]							     =  input_ids[-self.model_max_length :]
       return input_ids





    def 					lowerCAmelCase__					(      self	:							List[Any]	,  a	:							str	,  a	:							Optional[str] = None      ):






       '''simple docstring'''






       lowercase_						:     int							     =  0
       if os.path.isdir(a      ):
          lowercase_						:     Union[str, Any]							     =  os.path.join(
              a	,  (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]      )
          lowercase_						:     str							     =  os.path.join(
              a	,  (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["emoji_file"]      )
       else:
          lowercase_						:     Any							     =  (
              (filename_prefix + "-" if filename_prefix else "") + save_directory + VOCAB_FILES_NAMES["vocab_file"]
          )
          lowercase_						:     str							     =  (
              (filename_prefix + "-" if filename_prefix else "") + save_directory + VOCAB_FILES_NAMES["emoji_file"]
          )
       with open(a	,  "w"	,  encoding="utf-8"      ) as writer:
          for token_index, token in self.ids_to_tokens.items():
             if index != token_index:
                logger.warning(
                    f"""Saving vocabulary to {vocab_file}: vocabulary indices are not consecutive."""
                    " Please check that the vocabulary is not corrupted!"      )
                lowercase_						:     Union[str, Any]							     =  token_index
             writer.write(",".join(a      ) + "\n"      )
             index += 1
       with open(a	,  "w"	,  encoding="utf-8"      ) as writer:
          json.dump(self.emoji	,  a      )
       return vocab_file, emoji_file


class 							_UpperCAmelCase    (		snake_case      ):




    def __init__(      self	:							Dict	,  a	:							Union[str, Any]	,  a	:							List[str]	,  a	:							List[Any]      ):






       '''simple docstring'''






       lowercase_						:     Any							     =  vocab  # same as swe
       lowercase_						:     Union[str, Any]							     =  ids_to_tokens  # same as bpe
       lowercase_						:     str							     =  emoji
       lowercase_						:     Dict							     =  np.max([len(a      ) for w in self.vocab.keys()]      )
       lowercase_						:     Optional[int]							     =  re.compile(R"(https?|ftp)(:\/\/[-_\.!~*\'()a-zA-Z0-9;\/?:\@&=\+$,%#]+)"      )
       lowercase_						:     Tuple							     =  re.compile(R"[A-Za-z0-9\._+]*@[\-_0-9A-Za-z]+(\.[A-Za-z]+)*"      )
       lowercase_						:     List[Any]							     =  re.compile(R"[\(]{0,1}[0-9]{2,4}[\)\-\(]{0,1}[0-9]{2,4}[\)\-]{0,1}[0-9]{3,4}"      )
       lowercase_						:     Any							     =  re.compile(
           R"([12]\d{3}[/\-年])*(0?[1-9]|1[0-2])[/\-月]((0?[1-9]|[12][0-9]|3[01])日?)*(\d{1,2}|:|\d{1,2}時|\d{1,2}分|\(日\)|\(月\)|\(火\)|\(水\)|\(木\)|\(金\)|\(土\)|㈰|㈪|㈫|㈬|㈭|㈮|㈯)*"      )
       lowercase_						:     Tuple							     =  re.compile(
           R"(明治|大正|昭和|平成|令和|㍾|㍽|㍼|㍻|\u32ff)\d{1,2}年(0?[1-9]|1[0-2])月(0?[1-9]|[12][0-9]|3[01])日(\d{1,2}|:|\d{1,2}時|\d{1,2}分|\(日\)|\(月\)|\(火\)|\(水\)|\(木\)|\(金\)|\(土\)|㈰|㈪|㈫|㈬|㈭|㈮|㈯)*"      )
       lowercase_						:     List[Any]							     =  re.compile(
           R"((0|[1-9]\d*|[1-9]\d{0,2}(,\d{3})+)*億)*((0|[1-9]\d*|[1-9]\d{0,2}(,\d{3})+)*万)*((0|[1-9]\d*|[1-9]\d{0,2}(,\d{3})+)*千)*(0|[1-9]\d*|[1-9]\d{0,2}(,\d{3})+)*(千円|万円|千万円|円|千ドル|万ドル|千万ドル|ドル|千ユーロ|万ユーロ|千万ユーロ|ユーロ)+(\(税込\)|\(税抜\)|\+tax)*"      )
       lowercase_						:     Optional[Any]							     =  "─━│┃┄┅┆┇┈┉┊┋┌┍┎┏┐┑┒┓└┕┖┗┘┙┚┛├┝┞┟┠┡┢┣┤┥┦┧┨┩┪┫┬┭┮┯┰┱┲┳┴┵┶┷┸┹┺┻┼┽┾┿╀╁╂╃╄╅╆╇╈╉╊╋╌╍╎╏═║╒╓╔╕╖╗╘╙╚╛╜╝╞╟╠╡╢╣╤╥╦╧╨╩╪╫╬╭╮╯╰╱╲╳╴╵╶╷╸╹╺╻╼╽╾╿"
       lowercase_						:     List[str]							     =  "▀▁▂▃▄▅▆▇█▉▊▋▌▍▎▏▐░▒▓▔▕▖▗▘▙▚▛▜▝▞▟"
       lowercase_						:     Optional[Any]							     =  str.maketrans({k: "<BLOCK>" for k in keisen + blocks}      )




    def __len__(      self	:							List[Any]      ):






       '''simple docstring'''






       return len(self.ids_to_tokens      )




    def 					lowerCAmelCase__					(      self	:							Union[str, Any]	,  a	:							Union[str, Any]      ):






       '''simple docstring'''






       lowercase_						:     Union[str, Any]							     =  self.content_repattera.sub("<URL>"	,  a      )
       lowercase_						:     Tuple							     =  self.content_repattera.sub("<EMAIL>"	,  a      )
       lowercase_						:     List[str]							     =  self.content_repattera.sub("<TEL>"	,  a      )
       lowercase_						:     Any							     =  self.content_repattera.sub("<DATE>"	,  a      )
       lowercase_						:     int							     =  self.content_repattera.sub("<DATE>"	,  a      )
       lowercase_						:     Dict							     =  self.content_repattera.sub("<PRICE>"	,  a      )
       lowercase_						:     Any							     =  content.translate(self.content_transa      )
       while "<BLOCK><BLOCK>" in content:
          lowercase_						:     List[str]							     =  content.replace("<BLOCK><BLOCK>"	,  "<BLOCK>"      )
       return content




    def 					lowerCAmelCase__					(      self	:							List[str]	,  a	:							str	,  a	:							Optional[Any]=False      ):






       '''simple docstring'''






       lowercase_						:     List[str]							     =  text.replace(" "	,  "<SP>"      )
       lowercase_						:     Union[str, Any]							     =  text.replace("　"	,  "<SP>"      )
       lowercase_						:     int							     =  text.replace("\r\n"	,  "<BR>"      )
       lowercase_						:     Optional[Any]							     =  text.replace("\n"	,  "<BR>"      )
       lowercase_						:     Optional[Any]							     =  text.replace("\r"	,  "<BR>"      )
       lowercase_						:     str							     =  text.replace("\t"	,  "<TAB>"      )
       lowercase_						:     int							     =  text.replace("—"	,  "ー"      )
       lowercase_						:     int							     =  text.replace("−"	,  "ー"      )
       for k, v in self.emoji["emoji"].items():
          if k in text:
             lowercase_						:     Optional[int]							     =  text.replace(a	,  a      )
       if clean:
          lowercase_						:     List[str]							     =  self.clean_text(a      )

       def check_simbol(a	:							Union[str, Any]      ):
          lowercase_						:     List[str]							     =  x.encode()
          if len(a      ) == 1 and len(a      ) == 2:
             lowercase_						:     List[str]							     =  (int(e[0]      ) << 8) + int(e[1]      )
             if (
                 (c >= 0xC2A1 and c <= 0xC2BF)
                 or (c >= 0xC780 and c <= 0xC783)
                 or (c >= 0xCAB9 and c <= 0xCBBF)
                 or (c >= 0xCC80 and c <= 0xCDA2)
             ):
                return True
          return False

       def checkuae(a	:							Any      ):
          lowercase_						:     Tuple							     =  x.encode()
          if len(a      ) == 1 and len(a      ) == 3:
             lowercase_						:     Dict							     =  (int(e[0]      ) << 1_6) + (int(e[1]      ) << 8) + int(e[2]      )
             if c >= 0xE2_8080 and c <= 0xE2_B07F:
                return True
          return False

       lowercase_						:     List[Any]							     =  0
       lowercase_						:     List[str]							     =  []
       while pos < len(a      ):
          lowercase_						:     str							     =  min(len(a      )	,  pos + self.maxlen + 1      ) if text[pos] == "<" else pos + 3
          lowercase_						:     Optional[int]							     =  []  # (token_id, token, pos)
          for e in range(a	,  a	,  -1      ):
             lowercase_						:     Union[str, Any]							     =  text[pos:e]
             if wd in self.vocab:
                if wd[0] == "<" and len(a      ) > 2:
                   lowercase_						:     Tuple							     =  [(self.vocab[wd], wd, e)]
                   break
                else:
                   candidates.append((self.vocab[wd], wd, e)      )
          if len(a      ) > 0:
             # the smallest token_id is adopted
             lowercase_     ,		lowercase_     ,		lowercase_						:     Optional[Any]							     =  sorted(a	,  key=lambda a      : x[0]      )[0]
             result.append(a      )
             lowercase_						:     str							     =  e
          else:
             lowercase_						:     Optional[int]							     =  pos + 1
             lowercase_						:     Dict							     =  text[pos:end]
             if check_simbol(a      ):
                result.append("<KIGOU>"      )
             elif checkuae(a      ):
                result.append("<U2000U2BFF>"      )
             else:
                for i in wd.encode("utf-8"      ):
                   result.append("<|byte%d|>" % i      )
             lowercase_						:     Optional[Any]							     =  end
       return result





    def 					lowerCAmelCase__					(      self	:							Any	,  a	:							List[str]	,  a	:							List[Any]="\n"      ):






       '''simple docstring'''






       lowercase_						:     Optional[int]							     =  []
       lowercase_						:     str							     =  []
       lowercase_						:     Tuple							     =  self.ids_to_tokens[index][0]
       if word[:6] == "<|byte" and word[-2:] == "|>":
          byte_tokens.append(int(word[6:-2]      )      )
       else:
          if len(a      ) > 0:
             words.append(bytearray(a      ).decode("utf-8"	,  errors="replace"      )      )
             lowercase_						:     Union[str, Any]							     =  []
          if word[:7] == "<|emoji" and word[-2:] == "|>":
             words.append(self.emoji["emoji_inv"][word]      )
          elif word == "<SP>":
             words.append(" "      )
          elif word == "<BR>":
             words.append(a      )
          elif word == "<TAB>":
             words.append("\t"      )
          elif word == "<BLOCK>":
             words.append("▀"      )
          elif word == "<KIGOU>":
             words.append("ǀ"      )
          elif word == "<U2000U2BFF>":
             words.append("‖"      )
          else:
             words.append(a      )
       if len(a      ) > 0:
          words.append(bytearray(a      ).decode("utf-8"	,  errors="replace"      )      )
       lowercase_						:     Union[str, Any]							     =  "".join(a      )
       return text












'''simple docstring'''
import gc
import random
import unittest

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

import diffusers
from diffusers import (
    AutoencoderKL,
    EulerDiscreteScheduler,
    StableDiffusionLatentUpscalePipeline,
    StableDiffusionPipeline,
    UNetaDConditionModel,
)
from diffusers.schedulers import KarrasDiffusionSchedulers
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS, TEXT_GUIDED_IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()




def 		__SCREAMING_SNAKE_CASE       (	_UpperCamelCase				):


   """simple docstring"""




   lowercase_						:     Dict							     =  [tensor.shape for tensor in tensor_list]
   return all(shape == shapes[0] for shape in shapes[1:]				)


class 							_UpperCAmelCase    (		snake_case					,       snake_case					,       snake_case					,       unittest.TestCase      ):
    __lowerCamelCase: Union[str, Any]									=     StableDiffusionLatentUpscalePipeline
    __lowerCamelCase: List[Any]									=     TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {
        'height',
        'width',
        'cross_attention_kwargs',
        'negative_prompt_embeds',
        'prompt_embeds',
    }
    __lowerCamelCase: List[Any]									=     PipelineTesterMixin.required_optional_params - {'num_images_per_prompt'}
    __lowerCamelCase: List[Any]									=     TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
    __lowerCamelCase: Union[str, Any]									=     frozenset(
        []      )  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
    __lowerCamelCase: List[Any]									=     frozenset([]      )

    __lowerCamelCase: Dict									=     True




    @property
    def 					lowerCAmelCase__					(      self	:							List[Any]      ):






       '''simple docstring'''






       lowercase_						:     Optional[int]							     =  1
       lowercase_						:     Any							     =  4
       lowercase_						:     int							     =  (1_6, 1_6)

       lowercase_						:     List[Any]							     =  floats_tensor((batch_size, num_channels) + sizes	,  rng=random.Random(0      )      ).to(a      )
       return image




    def 					lowerCAmelCase__					(      self	:							Optional[Any]      ):






       '''simple docstring'''






       torch.manual_seed(0      )
       lowercase_						:     int							     =  UNetaDConditionModel(
           act_fn="gelu"	,  attention_head_dim=8	,  norm_num_groups=a	,  block_out_channels=[3_2, 3_2, 6_4, 6_4]	,  time_cond_proj_dim=1_6_0	,  conv_in_kernel=1	,  conv_out_kernel=1	,  cross_attention_dim=3_2	,  down_block_types=(
               "KDownBlock2D",
               "KCrossAttnDownBlock2D",
               "KCrossAttnDownBlock2D",
               "KCrossAttnDownBlock2D",
           )	,  in_channels=8	,  mid_block_type=a	,  only_cross_attention=a	,  out_channels=5	,  resnet_time_scale_shift="scale_shift"	,  time_embedding_type="fourier"	,  timestep_post_act="gelu"	,  up_block_types=("KCrossAttnUpBlock2D", "KCrossAttnUpBlock2D", "KCrossAttnUpBlock2D", "KUpBlock2D")	,  )
       lowercase_						:     List[str]							     =  AutoencoderKL(
           block_out_channels=[3_2, 3_2, 6_4, 6_4]	,  in_channels=3	,  out_channels=3	,  down_block_types=[
               "DownEncoderBlock2D",
               "DownEncoderBlock2D",
               "DownEncoderBlock2D",
               "DownEncoderBlock2D",
           ]	,  up_block_types=["UpDecoderBlock2D", "UpDecoderBlock2D", "UpDecoderBlock2D", "UpDecoderBlock2D"]	,  latent_channels=4	,  )
       lowercase_						:     Optional[int]							     =  EulerDiscreteScheduler(prediction_type="sample"      )
       lowercase_						:     List[str]							     =  CLIPTextConfig(
           bos_token_id=0	,  eos_token_id=2	,  hidden_size=3_2	,  intermediate_size=3_7	,  layer_norm_eps=1e-05	,  num_attention_heads=4	,  num_hidden_layers=5	,  pad_token_id=1	,  vocab_size=1_0_0_0	,  hidden_act="quick_gelu"	,  projection_dim=5_1_2	,  )
       lowercase_						:     Optional[int]							     =  CLIPTextModel(a      )
       lowercase_						:     List[str]							     =  CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"      )

       lowercase_						:     int							     =  {
           "unet": model.eval(),
           "vae": vae.eval(),
           "scheduler": scheduler,
           "text_encoder": text_encoder,
           "tokenizer": tokenizer,
       }

       return components




    def 					lowerCAmelCase__					(      self	:							List[str]	,  a	:							int	,  a	:							List[str]=0      ):






       '''simple docstring'''






       if str(a      ).startswith("mps"      ):
          lowercase_						:     Optional[int]							     =  torch.manual_seed(a      )
       else:
          lowercase_						:     Optional[int]							     =  torch.Generator(device=a      ).manual_seed(a      )
       lowercase_						:     Optional[Any]							     =  {
           "prompt": "A painting of a squirrel eating a burger",
           "image": self.dummy_image.cpu(),
           "generator": generator,
           "num_inference_steps": 2,
           "output_type": "numpy",
       }
       return inputs




    def 					lowerCAmelCase__					(      self	:							List[str]      ):






       '''simple docstring'''






       lowercase_						:     Optional[int]							     =  "cpu"

       lowercase_						:     Any							     =  self.get_dummy_components()
       lowercase_						:     Tuple							     =  self.pipeline_class(**a      )
       pipe.to(a      )
       pipe.set_progress_bar_config(disable=a      )

       lowercase_						:     List[Any]							     =  self.get_dummy_inputs(a      )
       lowercase_						:     List[str]							     =  pipe(**a      ).images
       lowercase_						:     int							     =  image[0, -3:, -3:, -1]

       self.assertEqual(image.shape	,  (1, 2_5_6, 2_5_6, 3)      )
       lowercase_						:     int							     =  np.array(
           [0.4722_2412, 0.4192_1633, 0.4471_7434, 0.4687_4192, 0.4258_8258, 0.4615_0726, 0.467_7534, 0.4558_3832, 0.4857_9055]      )
       lowercase_						:     int							     =  np.abs(image_slice.flatten() - expected_slice      ).max()
       self.assertLessEqual(a	,  1e-3      )




    def 					lowerCAmelCase__					(      self	:							int      ):






       '''simple docstring'''






       super().test_attention_slicing_forward_pass(expected_max_diff=7e-3      )




    def 					lowerCAmelCase__					(      self	:							List[Any]      ):






       '''simple docstring'''






       super().test_cpu_offload_forward_pass(expected_max_diff=3e-3      )




    def 					lowerCAmelCase__					(      self	:							List[str]      ):






       '''simple docstring'''






       super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3      )




    def 					lowerCAmelCase__					(      self	:							Dict      ):






       '''simple docstring'''






       super().test_inference_batch_single_identical(expected_max_diff=7e-3      )




    def 					lowerCAmelCase__					(      self	:							Union[str, Any]      ):






       '''simple docstring'''






       super().test_pt_np_pil_outputs_equivalent(expected_max_diff=3e-3      )




    def 					lowerCAmelCase__					(      self	:							Tuple      ):






       '''simple docstring'''






       super().test_save_load_local(expected_max_difference=3e-3      )




    def 					lowerCAmelCase__					(      self	:							Any      ):






       '''simple docstring'''






       super().test_save_load_optional_components(expected_max_difference=3e-3      )





    def 					lowerCAmelCase__					(      self	:							Optional[Any]      ):






       '''simple docstring'''






       lowercase_						:     Union[str, Any]							     =  [
           "DDIMScheduler",
           "DDPMScheduler",
           "PNDMScheduler",
           "HeunDiscreteScheduler",
           "EulerAncestralDiscreteScheduler",
           "KDPM2DiscreteScheduler",
           "KDPM2AncestralDiscreteScheduler",
           "DPMSolverSDEScheduler",
       ]
       lowercase_						:     Optional[int]							     =  self.get_dummy_components()
       lowercase_						:     int							     =  self.pipeline_class(**a      )

       # make sure that PNDM does not need warm-up
       pipe.scheduler.register_to_config(skip_prk_steps=a      )

       pipe.to(a      )
       pipe.set_progress_bar_config(disable=a      )
       lowercase_						:     Dict							     =  self.get_dummy_inputs(a      )
       lowercase_						:     int							     =  2

       lowercase_						:     Dict							     =  []
       for scheduler_enum in KarrasDiffusionSchedulers:
          if scheduler_enum.name in skip_schedulers:
             # no sigma schedulers are not supported
             # no schedulers
             continue

          lowercase_						:     Optional[int]							     =  getattr(a	,  scheduler_enum.name      )
          lowercase_						:     Union[str, Any]							     =  scheduler_cls.from_config(pipe.scheduler.config      )
          lowercase_						:     Tuple							     =  pipe(**a      )[0]
          outputs.append(a      )

       assert check_same_shape(a      )


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




    def 					lowerCAmelCase__					(      self	:							List[str]      ):






       '''simple docstring'''






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




    def 					lowerCAmelCase__					(      self	:							Tuple      ):






       '''simple docstring'''






       lowercase_						:     str							     =  torch.manual_seed(3_3      )

       lowercase_						:     Union[str, Any]							     =  StableDiffusionPipeline.from_pretrained("CompVis/stable-diffusion-v1-4"	,  torch_dtype=torch.floataa      )
       pipe.to("cuda"      )

       lowercase_						:     List[Any]							     =  StableDiffusionLatentUpscalePipeline.from_pretrained(
           "stabilityai/sd-x2-latent-upscaler"	,  torch_dtype=torch.floataa      )
       upscaler.to("cuda"      )

       lowercase_						:     Dict							     =  "a photo of an astronaut high resolution, unreal engine, ultra realistic"

       lowercase_						:     Optional[Any]							     =  pipe(a	,  generator=a	,  output_type="latent"      ).images

       lowercase_						:     Tuple							     =  upscaler(
           prompt=a	,  image=a	,  num_inference_steps=2_0	,  guidance_scale=0	,  generator=a	,  output_type="np"	,  ).images[0]

       lowercase_						:     Any							     =  load_numpy(
           "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/latent-upscaler/astronaut_1024.npy"      )
       assert np.abs((expected_image - image).mean()      ) < 5e-2





    def 					lowerCAmelCase__					(      self	:							Tuple      ):






       '''simple docstring'''






       lowercase_						:     List[Any]							     =  torch.manual_seed(3_3      )

       lowercase_						:     Optional[Any]							     =  StableDiffusionLatentUpscalePipeline.from_pretrained(
           "stabilityai/sd-x2-latent-upscaler"	,  torch_dtype=torch.floataa      )
       upscaler.to("cuda"      )

       lowercase_						:     Any							     =  "the temple of fire by Ross Tran and Gerardo Dottori, oil on canvas"

       lowercase_						:     str							     =  load_image(
           "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/latent-upscaler/fire_temple_512.png"      )

       lowercase_						:     int							     =  upscaler(
           prompt=a	,  image=a	,  num_inference_steps=2_0	,  guidance_scale=0	,  generator=a	,  output_type="np"	,  ).images[0]

       lowercase_						:     Union[str, Any]							     =  load_numpy(
           "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/latent-upscaler/fire_temple_1024.npy"      )
       assert np.abs((expected_image - image).max()      ) < 5e-2




