


import json
import os
from functools import lru_cache
from typing import TYPE_CHECKING, List, Optional, Tuple

import regex as re

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


if TYPE_CHECKING:
    from transformers.pipelines.conversational import Conversation

__UpperCamelCase    :						int     =							logging.get_logger(__name__)


__UpperCamelCase    :						Any     =							{
    'vocab_file': 'vocab.json',
    'merges_file': 'merges.txt',
    'tokenizer_config_file': 'tokenizer_config.json',
}

__UpperCamelCase    :						Dict     =							{
    'vocab_file': {'facebook/blenderbot-3B': 'https://huggingface.co/facebook/blenderbot-3B/resolve/main/vocab.json'},
    'merges_file': {'facebook/blenderbot-3B': 'https://huggingface.co/facebook/blenderbot-3B/resolve/main/merges.txt'},
    'tokenizer_config_file': {
        'facebook/blenderbot-3B': 'https://huggingface.co/facebook/blenderbot-3B/resolve/main/tokenizer_config.json'
    },
}

__UpperCamelCase    :						Optional[int]     =							{'facebook/blenderbot-3B': 128}





@lru_cache()
# Copied from transformers.models.roberta.tokenization_roberta.bytes_to_unicode
def   _UpperCAmelCase       (     ):
      """simple docstring"""




      __lowerCamelCase     :      Tuple       						=						(
          list(range(ord("""!"""      )						,	ord("""~"""      ) + 1      )      ) + list(range(ord("""¡"""      )						,	ord("""¬"""      ) + 1      )      ) + list(range(ord("""®"""      )						,	ord("""ÿ"""      ) + 1      )      )
      )
      __lowerCamelCase     :      Any       						=						bs[:]
      __lowerCamelCase     :      Tuple       						=						0
      for b in range(2**8      ):
            if b not in bs:
                  bs.append(UpperCAmelCase      )
                  cs.append(2**8 + n      )
                  n += 1
      __lowerCamelCase     :      int       						=						[chr(UpperCAmelCase      ) for n in cs]
      return dict(zip(UpperCAmelCase						,	UpperCAmelCase      )      )





def   _UpperCAmelCase       (     UpperCAmelCase      :			Dict      ):
      """simple docstring"""




      __lowerCamelCase     :      str       						=						set()
      __lowerCamelCase     :      Union[str, Any]       						=						word[0]
      for char in word[1:]:
            pairs.add((prev_char, char)      )
            __lowerCamelCase     :      Optional[int]       						=						char
      return pairs




class 							_UpperCamelCase (						A						):

   '''simple docstring'''







   a_				:							int    = VOCAB_FILES_NAMES
   a_				:							str    = PRETRAINED_VOCAB_FILES_MAP
   a_				:							int    = PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
   a_				:							List[str]    = ["input_ids", "attention_mask"]

   def __init__(  self  : List[Any]       ,  _lowerCamelCase  : Dict       ,  _lowerCamelCase  : int       ,  _lowerCamelCase  : List[str]="replace"       ,  _lowerCamelCase  : List[str]="<s>"       ,  _lowerCamelCase  : Any="</s>"       ,  _lowerCamelCase  : str="</s>"       ,  _lowerCamelCase  : Any="<s>"       ,  _lowerCamelCase  : Union[str, Any]="<unk>"       ,  _lowerCamelCase  : Union[str, Any]="<pad>"       ,  _lowerCamelCase  : Optional[int]="<mask>"       ,  _lowerCamelCase  : Tuple=False       ,  **_lowerCamelCase  : Dict       ,  ):






         '''simple docstring'''



         __lowerCamelCase     :      Any       						=						AddedToken(_lowerCamelCase       ,  lstrip=_lowerCamelCase       ,  rstrip=_lowerCamelCase							) if isinstance(_lowerCamelCase       ,  _lowerCamelCase							) else bos_token
         __lowerCamelCase     :      str       						=						AddedToken(_lowerCamelCase       ,  lstrip=_lowerCamelCase       ,  rstrip=_lowerCamelCase							) if isinstance(_lowerCamelCase       ,  _lowerCamelCase							) else eos_token
         __lowerCamelCase     :      str       						=						AddedToken(_lowerCamelCase       ,  lstrip=_lowerCamelCase       ,  rstrip=_lowerCamelCase							) if isinstance(_lowerCamelCase       ,  _lowerCamelCase							) else sep_token
         __lowerCamelCase     :      Tuple       						=						AddedToken(_lowerCamelCase       ,  lstrip=_lowerCamelCase       ,  rstrip=_lowerCamelCase							) if isinstance(_lowerCamelCase       ,  _lowerCamelCase							) else cls_token
         __lowerCamelCase     :      Tuple       						=						AddedToken(_lowerCamelCase       ,  lstrip=_lowerCamelCase       ,  rstrip=_lowerCamelCase							) if isinstance(_lowerCamelCase       ,  _lowerCamelCase							) else unk_token
         __lowerCamelCase     :      Dict       						=						AddedToken(_lowerCamelCase       ,  lstrip=_lowerCamelCase       ,  rstrip=_lowerCamelCase							) if isinstance(_lowerCamelCase       ,  _lowerCamelCase							) else pad_token

         # Mask token behave like a normal word, i.e. include the space before it
         __lowerCamelCase     :      Optional[int]       						=						AddedToken(_lowerCamelCase       ,  lstrip=_lowerCamelCase       ,  rstrip=_lowerCamelCase							) if isinstance(_lowerCamelCase       ,  _lowerCamelCase							) else mask_token

         super().__init__(
             errors=_lowerCamelCase       ,  bos_token=_lowerCamelCase       ,  eos_token=_lowerCamelCase       ,  unk_token=_lowerCamelCase       ,  sep_token=_lowerCamelCase       ,  cls_token=_lowerCamelCase       ,  pad_token=_lowerCamelCase       ,  mask_token=_lowerCamelCase       ,  add_prefix_space=_lowerCamelCase       ,  **_lowerCamelCase       ,  )

         with open(_lowerCamelCase       ,  encoding="""utf-8"""							) as vocab_handle:
               __lowerCamelCase     :      Tuple       						=						json.load(_lowerCamelCase							)
         __lowerCamelCase     :      Any       						=						{v: k for k, v in self.encoder.items()}
         __lowerCamelCase     :      Optional[Any]       						=						errors  # how to handle errors in decoding
         __lowerCamelCase     :      Optional[int]       						=						bytes_to_unicode()
         __lowerCamelCase     :      Optional[int]       						=						{v: k for k, v in self.byte_encoder.items()}
         with open(_lowerCamelCase       ,  encoding="""utf-8"""							) as merges_handle:
               __lowerCamelCase     :      int       						=						merges_handle.read().split("""\n"""							)[1:-1]
         __lowerCamelCase     :      Union[str, Any]       						=						[tuple(merge.split()							) for merge in bpe_merges]
         __lowerCamelCase     :      List[Any]       						=						dict(zip(_lowerCamelCase       ,  range(len(_lowerCamelCase							)							)							)							)
         __lowerCamelCase     :      str       						=						{}
         __lowerCamelCase     :      Optional[Any]       						=						add_prefix_space

         # Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
         __lowerCamelCase     :      Tuple       						=						re.compile(R"""'s|'t|'re|'ve|'m|'ll|'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+"""							)

   @property
   # Copied from transformers.models.roberta.tokenization_roberta.RobertaTokenizer.vocab_size with Roberta->Blenderbot, RoBERTa->Blenderbot
   def   _snake_case    (  self  : Any							):






         '''simple docstring'''



         return len(self.encoder							)

   def   _snake_case    (  self  : Optional[int]							):






         '''simple docstring'''



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

   def   _snake_case    (  self  : List[str]       ,  _lowerCamelCase  : Optional[int]							):






         '''simple docstring'''



         if token in self.cache:
               return self.cache[token]
         __lowerCamelCase     :      Dict       						=						tuple(_lowerCamelCase							)
         __lowerCamelCase     :      Dict       						=						get_pairs(_lowerCamelCase							)

         if not pairs:
               return token

         while True:
               __lowerCamelCase     :      List[Any]       						=						min(_lowerCamelCase       ,  key=lambda _lowerCamelCase							: self.bpe_ranks.get(_lowerCamelCase       ,  float("""inf"""							)							)							)
               if bigram not in self.bpe_ranks:
                     break
               __lowerCamelCase						, __lowerCamelCase     :      str       						=						bigram
               __lowerCamelCase     :      Tuple       						=						[]
               __lowerCamelCase     :      List[str]       						=						0
               while i < len(_lowerCamelCase							):
                     try:
                           __lowerCamelCase     :      List[str]       						=						word.index(_lowerCamelCase       ,  _lowerCamelCase							)
                     except ValueError:
                           new_word.extend(word[i:]							)
                           break
                     else:
                           new_word.extend(word[i:j]							)
                           __lowerCamelCase     :      int       						=						j

                     if word[i] == first and i < len(_lowerCamelCase							) - 1 and word[i + 1] == second:
                           new_word.append(first + second							)
                           i += 2
                     else:
                           new_word.append(word[i]							)
                           i += 1
               __lowerCamelCase     :      List[str]       						=						tuple(_lowerCamelCase							)
               __lowerCamelCase     :      List[Any]       						=						new_word
               if len(_lowerCamelCase							) == 1:
                     break
               else:
                     __lowerCamelCase     :      List[Any]       						=						get_pairs(_lowerCamelCase							)
         __lowerCamelCase     :      List[Any]       						=						""" """.join(_lowerCamelCase							)
         __lowerCamelCase     :      List[Any]       						=						word
         return word

   def   _snake_case    (  self  : int       ,  _lowerCamelCase  : List[str]							):






         '''simple docstring'''



         __lowerCamelCase     :      Optional[int]       						=						[]
         for token in re.findall(self.pat       ,  _lowerCamelCase							):
               __lowerCamelCase     :      Tuple       						=						"""""".join(
                   self.byte_encoder[b] for b in token.encode("""utf-8"""							)							)  # Maps all our bytes to unicode strings, avoiding control tokens of the BPE (spaces in our case)
               bpe_tokens.extend(bpe_token for bpe_token in self.bpe(_lowerCamelCase							).split(""" """							)							)
         return bpe_tokens

   def   _snake_case    (  self  : str       ,  _lowerCamelCase  : Dict							):






         '''simple docstring'''



         return self.encoder.get(_lowerCamelCase       ,  self.encoder.get(self.unk_token							)							)

   def   _snake_case    (  self  : str       ,  _lowerCamelCase  : List[str]							):






         '''simple docstring'''



         return self.decoder.get(_lowerCamelCase							)

   def   _snake_case    (  self  : Optional[int]       ,  _lowerCamelCase  : Optional[int]							):






         '''simple docstring'''



         __lowerCamelCase     :      Union[str, Any]       						=						"""""".join(_lowerCamelCase							)
         __lowerCamelCase     :      Tuple       						=						bytearray([self.byte_decoder[c] for c in text]							).decode("""utf-8"""       ,  errors=self.errors							)
         return text

   def   _snake_case    (  self  : Optional[Any]       ,  _lowerCamelCase  : str       ,  _lowerCamelCase  : Optional[str] = None							):






         '''simple docstring'''



         if not os.path.isdir(_lowerCamelCase							):
               logger.error(F"""Vocabulary path ({save_directory}) should be a directory"""							)
               return
         __lowerCamelCase     :      Tuple       						=						os.path.join(
             _lowerCamelCase       ,  (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]							)
         __lowerCamelCase     :      List[Any]       						=						os.path.join(
             _lowerCamelCase       ,  (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""merges_file"""]							)

         with open(_lowerCamelCase       ,  """w"""       ,  encoding="""utf-8"""							) as f:
               f.write(json.dumps(self.encoder       ,  indent=2       ,  sort_keys=_lowerCamelCase       ,  ensure_ascii=_lowerCamelCase							) + """\n"""							)

         __lowerCamelCase     :      str       						=						0
         with open(_lowerCamelCase       ,  """w"""       ,  encoding="""utf-8"""							) as writer:
               writer.write("""#version: 0.2\n"""							)
               for bpe_tokens, token_index in sorted(self.bpe_ranks.items()       ,  key=lambda _lowerCamelCase							: kv[1]							):
                     if index != token_index:
                           logger.warning(
                               F"""Saving vocabulary to {merge_file}: BPE merge indices are not consecutive."""
                               """ Please check that the tokenizer is not corrupted!"""							)
                           __lowerCamelCase     :      Any       						=						token_index
                     writer.write(""" """.join(_lowerCamelCase							) + """\n"""							)
                     index += 1

         return vocab_file, merge_file

   def   _snake_case    (  self  : Tuple       ,  _lowerCamelCase  : List[int]       ,  _lowerCamelCase  : Optional[List[int]] = None       ,  _lowerCamelCase  : bool = 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							)

         if token_ids_a is None:
               return [1] + ([0] * len(_lowerCamelCase							)) + [1]
         return [1] + ([0] * len(_lowerCamelCase							)) + [1, 1] + ([0] * len(_lowerCamelCase							)) + [1]

   def   _snake_case    (  self  : int       ,  _lowerCamelCase  : List[int]       ,  _lowerCamelCase  : Optional[List[int]] = None							):






         '''simple docstring'''



         __lowerCamelCase     :      Dict       						=						[self.sep_token_id]
         __lowerCamelCase     :      Any       						=						[self.cls_token_id]

         if token_ids_a is None:
               return len(cls + token_ids_a + sep							) * [0]
         return len(cls + token_ids_a + sep + sep + token_ids_a + sep							) * [0]

   def   _snake_case    (  self  : Union[str, Any]       ,  _lowerCamelCase  : Dict       ,  _lowerCamelCase  : Union[str, Any]=False       ,  **_lowerCamelCase  : int							):






         '''simple docstring'''



         __lowerCamelCase     :      Any       						=						kwargs.pop("""add_prefix_space"""       ,  self.add_prefix_space							)
         if (is_split_into_words or add_prefix_space) and (len(_lowerCamelCase							) > 0 and not text[0].isspace()):
               __lowerCamelCase     :      Any       						=						""" """ + text
         return (text, kwargs)

   def   _snake_case    (  self  : List[Any]       ,  _lowerCamelCase  : List[int]       ,  _lowerCamelCase  : Optional[List[int]] = None							):






         '''simple docstring'''



         return token_ids_a + [self.eos_token_id]


   def   _snake_case    (  self  : Union[str, Any]       ,  _lowerCamelCase  : "Conversation"							):






         '''simple docstring'''



         __lowerCamelCase     :      Union[str, Any]       						=						[]
         for is_user, text in conversation.iter_texts():
               if is_user:
                     # We need to space prefix as it's being done within blenderbot
                     inputs.append(""" """ + text							)
               else:
                     # Generated responses should contain them already.
                     inputs.append(_lowerCamelCase							)

         __lowerCamelCase     :      Optional[int]       						=						"""  """.join(_lowerCamelCase							)
         __lowerCamelCase     :      Dict       						=						self.encode(_lowerCamelCase							)
         if len(_lowerCamelCase							) > self.model_max_length:
               __lowerCamelCase     :      Union[str, Any]       						=						input_ids[-self.model_max_length :]
               logger.warning(F"""Trimmed input from conversation as it was longer than {self.model_max_length} tokens."""							)
         return input_ids







import random
import unittest

import numpy as np
import torch

from diffusers import (
    DPMSolverMultistepScheduler,
    EulerAncestralDiscreteScheduler,
    EulerDiscreteScheduler,
    LMSDiscreteScheduler,
    OnnxStableDiffusionUpscalePipeline,
    PNDMScheduler,
)
from diffusers.utils import floats_tensor
from diffusers.utils.testing_utils import (
    is_onnx_available,
    load_image,
    nightly,
    require_onnxruntime,
    require_torch_gpu,
)

from ..test_pipelines_onnx_common import OnnxPipelineTesterMixin


if is_onnx_available():
    import onnxruntime as ort



class 							_UpperCamelCase (						A,unittest.TestCase						):

   '''simple docstring'''







   a_				:							Any    = "ssube/stable-diffusion-x4-upscaler-onnx"

   def   _snake_case    (  self  : Any       ,  _lowerCamelCase  : List[str]=0							):






         '''simple docstring'''



         __lowerCamelCase     :      Union[str, Any]       						=						floats_tensor((1, 3, 1_2_8, 1_2_8)       ,  rng=random.Random(_lowerCamelCase							)							)
         __lowerCamelCase     :      int       						=						torch.manual_seed(_lowerCamelCase							)
         __lowerCamelCase     :      Union[str, Any]       						=						{
             """prompt""": """A painting of a squirrel eating a burger""",
             """image""": image,
             """generator""": generator,
             """num_inference_steps""": 3,
             """guidance_scale""": 7.5,
             """output_type""": """numpy""",
         }
         return inputs

   def   _snake_case    (  self  : Optional[Any]							):






         '''simple docstring'''



         __lowerCamelCase     :      Union[str, Any]       						=						OnnxStableDiffusionUpscalePipeline.from_pretrained(self.hub_checkpoint       ,  provider="""CPUExecutionProvider"""							)
         pipe.set_progress_bar_config(disable=_lowerCamelCase							)

         __lowerCamelCase     :      Tuple       						=						self.get_dummy_inputs()
         __lowerCamelCase     :      Dict       						=						pipe(**_lowerCamelCase							).images
         __lowerCamelCase     :      Dict       						=						image[0, -3:, -3:, -1].flatten()

         # started as 128, should now be 512
         assert image.shape == (1, 5_1_2, 5_1_2, 3)
         __lowerCamelCase     :      str       						=						np.array(
             [0.6_974_782, 0.68_902_093, 0.70_135_885, 0.7_583_618, 0.7_804_545, 0.7_854_912, 0.78_667_426, 0.78_743_863, 0.78_070_223]							)
         assert np.abs(image_slice - expected_slice							).max() < 1E-1

   def   _snake_case    (  self  : Dict							):






         '''simple docstring'''



         __lowerCamelCase     :      Optional[Any]       						=						OnnxStableDiffusionUpscalePipeline.from_pretrained(self.hub_checkpoint       ,  provider="""CPUExecutionProvider"""							)
         __lowerCamelCase     :      str       						=						PNDMScheduler.from_config(pipe.scheduler.config       ,  skip_prk_steps=_lowerCamelCase							)
         pipe.set_progress_bar_config(disable=_lowerCamelCase							)

         __lowerCamelCase     :      Tuple       						=						self.get_dummy_inputs()
         __lowerCamelCase     :      Union[str, Any]       						=						pipe(**_lowerCamelCase							).images
         __lowerCamelCase     :      List[Any]       						=						image[0, -3:, -3:, -1]

         assert image.shape == (1, 5_1_2, 5_1_2, 3)
         __lowerCamelCase     :      int       						=						np.array(
             [0.6_898_892, 0.59_240_556, 0.52_499_527, 0.58_866_215, 0.52_258_235, 0.52_572_715, 0.62_414_473, 0.6_174_387, 0.6_214_964]							)
         assert np.abs(image_slice.flatten() - expected_slice							).max() < 1E-1

   def   _snake_case    (  self  : Tuple							):






         '''simple docstring'''



         __lowerCamelCase     :      Optional[int]       						=						OnnxStableDiffusionUpscalePipeline.from_pretrained(self.hub_checkpoint       ,  provider="""CPUExecutionProvider"""							)
         __lowerCamelCase     :      int       						=						DPMSolverMultistepScheduler.from_config(pipe.scheduler.config							)
         pipe.set_progress_bar_config(disable=_lowerCamelCase							)

         __lowerCamelCase     :      List[str]       						=						self.get_dummy_inputs()
         __lowerCamelCase     :      List[str]       						=						pipe(**_lowerCamelCase							).images
         __lowerCamelCase     :      Optional[Any]       						=						image[0, -3:, -3:, -1]

         assert image.shape == (1, 5_1_2, 5_1_2, 3)
         __lowerCamelCase     :      str       						=						np.array(
             [0.7_659_278, 0.76_437_664, 0.75_579_107, 0.7_691_116, 0.77_666_986, 0.7_727_672, 0.7_758_664, 0.7_812_226, 0.76_942_515]							)

         assert np.abs(image_slice.flatten() - expected_slice							).max() < 1E-1

   def   _snake_case    (  self  : str							):






         '''simple docstring'''



         __lowerCamelCase     :      Optional[int]       						=						OnnxStableDiffusionUpscalePipeline.from_pretrained(self.hub_checkpoint       ,  provider="""CPUExecutionProvider"""							)
         __lowerCamelCase     :      Optional[Any]       						=						EulerDiscreteScheduler.from_config(pipe.scheduler.config							)
         pipe.set_progress_bar_config(disable=_lowerCamelCase							)

         __lowerCamelCase     :      Dict       						=						self.get_dummy_inputs()
         __lowerCamelCase     :      Dict       						=						pipe(**_lowerCamelCase							).images
         __lowerCamelCase     :      List[str]       						=						image[0, -3:, -3:, -1]

         assert image.shape == (1, 5_1_2, 5_1_2, 3)
         __lowerCamelCase     :      List[str]       						=						np.array(
             [0.6_974_782, 0.68_902_093, 0.70_135_885, 0.7_583_618, 0.7_804_545, 0.7_854_912, 0.78_667_426, 0.78_743_863, 0.78_070_223]							)
         assert np.abs(image_slice.flatten() - expected_slice							).max() < 1E-1


   def   _snake_case    (  self  : Any							):






         '''simple docstring'''



         __lowerCamelCase     :      List[str]       						=						OnnxStableDiffusionUpscalePipeline.from_pretrained(self.hub_checkpoint       ,  provider="""CPUExecutionProvider"""							)
         __lowerCamelCase     :      Union[str, Any]       						=						EulerAncestralDiscreteScheduler.from_config(pipe.scheduler.config							)
         pipe.set_progress_bar_config(disable=_lowerCamelCase							)

         __lowerCamelCase     :      Optional[int]       						=						self.get_dummy_inputs()
         __lowerCamelCase     :      int       						=						pipe(**_lowerCamelCase							).images
         __lowerCamelCase     :      int       						=						image[0, -3:, -3:, -1]

         assert image.shape == (1, 5_1_2, 5_1_2, 3)
         __lowerCamelCase     :      Optional[int]       						=						np.array(
             [0.77_424_496, 0.773_601, 0.7_645_288, 0.7_769_598, 0.7_772_739, 0.7_738_688, 0.78_187_233, 0.77_879_584, 0.767_043]							)

         assert np.abs(image_slice.flatten() - expected_slice							).max() < 1E-1



@nightly
@require_onnxruntime
@require_torch_gpu
class 							_UpperCamelCase (						unittest.TestCase						):

   '''simple docstring'''







   @property
   def   _snake_case    (  self  : str							):






         '''simple docstring'''



         return (
             "CUDAExecutionProvider",
             {
                 "gpu_mem_limit": "15000000000",  # 15GB
                 "arena_extend_strategy": "kSameAsRequested",
             },
         )

   @property
   def   _snake_case    (  self  : Tuple							):






         '''simple docstring'''



         __lowerCamelCase     :      Any       						=						ort.SessionOptions()
         __lowerCamelCase     :      str       						=						False
         return options

   def   _snake_case    (  self  : Any							):






         '''simple docstring'''



         __lowerCamelCase     :      str       						=						load_image(
             """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
             """/img2img/sketch-mountains-input.jpg"""							)
         __lowerCamelCase     :      Optional[Any]       						=						init_image.resize((1_2_8, 1_2_8)							)
         # using the PNDM scheduler by default
         __lowerCamelCase     :      List[Any]       						=						OnnxStableDiffusionUpscalePipeline.from_pretrained(
             """ssube/stable-diffusion-x4-upscaler-onnx"""       ,  provider=self.gpu_provider       ,  sess_options=self.gpu_options       ,  )
         pipe.set_progress_bar_config(disable=_lowerCamelCase							)

         __lowerCamelCase     :      List[Any]       						=						"""A fantasy landscape, trending on artstation"""

         __lowerCamelCase     :      str       						=						torch.manual_seed(0							)
         __lowerCamelCase     :      Any       						=						pipe(
             prompt=_lowerCamelCase       ,  image=_lowerCamelCase       ,  guidance_scale=7.5       ,  num_inference_steps=1_0       ,  generator=_lowerCamelCase       ,  output_type="""np"""       ,  )
         __lowerCamelCase     :      List[str]       						=						output.images
         __lowerCamelCase     :      Any       						=						images[0, 2_5_5:2_5_8, 3_8_3:3_8_6, -1]

         assert images.shape == (1, 5_1_2, 5_1_2, 3)
         __lowerCamelCase     :      int       						=						np.array([0.4_883, 0.4_947, 0.4_980, 0.4_975, 0.4_982, 0.4_980, 0.5_000, 0.5_006, 0.4_972]							)
         # TODO: lower the tolerance after finding the cause of onnxruntime reproducibility issues

         assert np.abs(image_slice.flatten() - expected_slice							).max() < 2E-2


   def   _snake_case    (  self  : Any							):






         '''simple docstring'''



         __lowerCamelCase     :      Tuple       						=						load_image(
             """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
             """/img2img/sketch-mountains-input.jpg"""							)
         __lowerCamelCase     :      Union[str, Any]       						=						init_image.resize((1_2_8, 1_2_8)							)
         __lowerCamelCase     :      str       						=						LMSDiscreteScheduler.from_pretrained(
             """ssube/stable-diffusion-x4-upscaler-onnx"""       ,  subfolder="""scheduler"""							)
         __lowerCamelCase     :      str       						=						OnnxStableDiffusionUpscalePipeline.from_pretrained(
             """ssube/stable-diffusion-x4-upscaler-onnx"""       ,  scheduler=_lowerCamelCase       ,  provider=self.gpu_provider       ,  sess_options=self.gpu_options       ,  )
         pipe.set_progress_bar_config(disable=_lowerCamelCase							)

         __lowerCamelCase     :      str       						=						"""A fantasy landscape, trending on artstation"""

         __lowerCamelCase     :      Tuple       						=						torch.manual_seed(0							)
         __lowerCamelCase     :      Dict       						=						pipe(
             prompt=_lowerCamelCase       ,  image=_lowerCamelCase       ,  guidance_scale=7.5       ,  num_inference_steps=2_0       ,  generator=_lowerCamelCase       ,  output_type="""np"""       ,  )
         __lowerCamelCase     :      int       						=						output.images
         __lowerCamelCase     :      List[Any]       						=						images[0, 2_5_5:2_5_8, 3_8_3:3_8_6, -1]

         assert images.shape == (1, 5_1_2, 5_1_2, 3)
         __lowerCamelCase     :      List[str]       						=						np.array(
             [0.50_173_753, 0.50_223_356, 0.502_039, 0.50_233_036, 0.5_023_725, 0.5_022_601, 0.5_018_758, 0.50_234_085, 0.50_241_566]							)
         # TODO: lower the tolerance after finding the cause of onnxruntime reproducibility issues

         assert np.abs(image_slice.flatten() - expected_slice							).max() < 2E-2



