


import os
import sys
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from huggingface_hub import HfFolder, delete_repo
from huggingface_hub.file_download import http_get
from requests.exceptions import HTTPError

from transformers import (
    AlbertTokenizer,
    AutoTokenizer,
    BertTokenizer,
    BertTokenizerFast,
    GPTaTokenizerFast,
    is_tokenizers_available,
)
from transformers.testing_utils import TOKEN, USER, is_staging_test, require_tokenizers
from transformers.tokenization_utils import Trie


sys.path.append(str(Path(__file__).parent.parent / 'utils'))

from test_module.custom_tokenization import CustomTokenizer  # noqa E402


if is_tokenizers_available():
    from test_module.custom_tokenization_fast import CustomTokenizerFast







class 							_UpperCamelCase (						unittest.TestCase						):

   '''simple docstring'''







   def   _snake_case    (  self  : Optional[Any]							):






         '''simple docstring'''



         __lowerCamelCase     :      Tuple       						=						mock.Mock()
         __lowerCamelCase     :      Optional[Any]       						=						5_0_0
         __lowerCamelCase     :      Optional[Any]       						=						{}
         __lowerCamelCase     :      List[str]       						=						HTTPError
         __lowerCamelCase     :      List[str]       						=						{}

         # Download this model to make sure it's in the cache.
         __lowerCamelCase     :      Optional[Any]       						=						BertTokenizer.from_pretrained("""hf-internal-testing/tiny-random-bert"""							)

         # Under the mock environment we get a 500 error when trying to reach the tokenizer.
         with mock.patch("""requests.Session.request"""       ,  return_value=_lowerCamelCase							) as mock_head:
               __lowerCamelCase     :      Optional[int]       						=						BertTokenizer.from_pretrained("""hf-internal-testing/tiny-random-bert"""							)
               # This check we did call the fake head request
               mock_head.assert_called()

   @require_tokenizers
   def   _snake_case    (  self  : Any							):






         '''simple docstring'''



         __lowerCamelCase     :      int       						=						mock.Mock()
         __lowerCamelCase     :      Dict       						=						5_0_0
         __lowerCamelCase     :      List[Any]       						=						{}
         __lowerCamelCase     :      Union[str, Any]       						=						HTTPError
         __lowerCamelCase     :      str       						=						{}

         # Download this model to make sure it's in the cache.
         __lowerCamelCase     :      Optional[Any]       						=						GPTaTokenizerFast.from_pretrained("""gpt2"""							)

         # Under the mock environment we get a 500 error when trying to reach the tokenizer.
         with mock.patch("""requests.Session.request"""       ,  return_value=_lowerCamelCase							) as mock_head:
               __lowerCamelCase     :      Any       						=						GPTaTokenizerFast.from_pretrained("""gpt2"""							)
               # This check we did call the fake head request
               mock_head.assert_called()

   def   _snake_case    (  self  : Dict							):






         '''simple docstring'''



         try:
               __lowerCamelCase     :      int       						=						tempfile.mktemp()
               with open(_lowerCamelCase       ,  """wb"""							) as f:
                     http_get("""https://huggingface.co/albert-base-v1/resolve/main/spiece.model"""       ,  _lowerCamelCase							)

               __lowerCamelCase     :      Optional[int]       						=						AlbertTokenizer.from_pretrained(_lowerCamelCase							)
         finally:
               os.remove(_lowerCamelCase							)

         # Supporting this legacy load introduced a weird bug where the tokenizer would load local files if they are in
         # the current folder and have the right name.
         if os.path.isfile("""tokenizer.json"""							):
               # We skip the test if the user has a `tokenizer.json` in this folder to avoid deleting it.
               return
         try:
               with open("""tokenizer.json"""       ,  """wb"""							) as f:
                     http_get("""https://huggingface.co/hf-internal-testing/tiny-random-bert/blob/main/tokenizer.json"""       ,  _lowerCamelCase							)
               __lowerCamelCase     :      Tuple       						=						AutoTokenizer.from_pretrained("""hf-internal-testing/tiny-random-gpt2"""							)
               # The tiny random BERT has a vocab size of 1024, tiny gpt2 as a vocab size of 1000
               self.assertEqual(tokenizer.vocab_size       ,  1_0_0_0							)
               # Tokenizer should depend on the remote checkpoint, not the local tokenizer.json file.

         finally:
               os.remove("""tokenizer.json"""							)


   def   _snake_case    (  self  : Optional[Any]							):






         '''simple docstring'''



         __lowerCamelCase     :      Dict       						=						AlbertTokenizer.from_pretrained("""https://huggingface.co/albert-base-v1/resolve/main/spiece.model"""							)







@is_staging_test
class 							_UpperCamelCase (						unittest.TestCase						):

   '''simple docstring'''







   a_				:							Union[str, Any]    = ["[UNK]", "[CLS]", "[SEP]", "[PAD]", "[MASK]", "bla", "blou"]

   @classmethod
   def   _snake_case    (  cls  : Any							):






         '''simple docstring'''



         __lowerCamelCase     :      Any       						=						TOKEN
         HfFolder.save_token(_lowerCamelCase							)

   @classmethod
   def   _snake_case    (  cls  : Optional[Any]							):






         '''simple docstring'''



         try:
               delete_repo(token=cls._token       ,  repo_id="""test-tokenizer"""							)
         except HTTPError:
               pass

         try:
               delete_repo(token=cls._token       ,  repo_id="""valid_org/test-tokenizer-org"""							)
         except HTTPError:
               pass

         try:
               delete_repo(token=cls._token       ,  repo_id="""test-dynamic-tokenizer"""							)
         except HTTPError:
               pass

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






         '''simple docstring'''



         with tempfile.TemporaryDirectory() as tmp_dir:
               __lowerCamelCase     :      Union[str, Any]       						=						os.path.join(_lowerCamelCase       ,  """vocab.txt"""							)
               with open(_lowerCamelCase       ,  """w"""       ,  encoding="""utf-8"""							) as vocab_writer:
                     vocab_writer.write("""""".join([x + """\n""" for x in self.vocab_tokens]							)							)
               __lowerCamelCase     :      Optional[Any]       						=						BertTokenizer(_lowerCamelCase							)

         tokenizer.push_to_hub("""test-tokenizer"""       ,  use_auth_token=self._token							)
         __lowerCamelCase     :      Any       						=						BertTokenizer.from_pretrained(F"""{USER}/test-tokenizer"""							)
         self.assertDictEqual(new_tokenizer.vocab       ,  tokenizer.vocab							)

         # Reset repo
         delete_repo(token=self._token       ,  repo_id="""test-tokenizer"""							)

         # Push to hub via save_pretrained
         with tempfile.TemporaryDirectory() as tmp_dir:
               tokenizer.save_pretrained(_lowerCamelCase       ,  repo_id="""test-tokenizer"""       ,  push_to_hub=_lowerCamelCase       ,  use_auth_token=self._token							)

         __lowerCamelCase     :      Any       						=						BertTokenizer.from_pretrained(F"""{USER}/test-tokenizer"""							)
         self.assertDictEqual(new_tokenizer.vocab       ,  tokenizer.vocab							)

   def   _snake_case    (  self  : List[str]							):






         '''simple docstring'''



         with tempfile.TemporaryDirectory() as tmp_dir:
               __lowerCamelCase     :      str       						=						os.path.join(_lowerCamelCase       ,  """vocab.txt"""							)
               with open(_lowerCamelCase       ,  """w"""       ,  encoding="""utf-8"""							) as vocab_writer:
                     vocab_writer.write("""""".join([x + """\n""" for x in self.vocab_tokens]							)							)
               __lowerCamelCase     :      str       						=						BertTokenizer(_lowerCamelCase							)

         tokenizer.push_to_hub("""valid_org/test-tokenizer-org"""       ,  use_auth_token=self._token							)
         __lowerCamelCase     :      List[Any]       						=						BertTokenizer.from_pretrained("""valid_org/test-tokenizer-org"""							)
         self.assertDictEqual(new_tokenizer.vocab       ,  tokenizer.vocab							)

         # Reset repo
         delete_repo(token=self._token       ,  repo_id="""valid_org/test-tokenizer-org"""							)

         # Push to hub via save_pretrained
         with tempfile.TemporaryDirectory() as tmp_dir:
               tokenizer.save_pretrained(
                   _lowerCamelCase       ,  repo_id="""valid_org/test-tokenizer-org"""       ,  push_to_hub=_lowerCamelCase       ,  use_auth_token=self._token							)

         __lowerCamelCase     :      Optional[int]       						=						BertTokenizer.from_pretrained("""valid_org/test-tokenizer-org"""							)
         self.assertDictEqual(new_tokenizer.vocab       ,  tokenizer.vocab							)


   @require_tokenizers
   def   _snake_case    (  self  : Optional[Any]							):






         '''simple docstring'''



         CustomTokenizer.register_for_auto_class()
         with tempfile.TemporaryDirectory() as tmp_dir:
               __lowerCamelCase     :      List[Any]       						=						os.path.join(_lowerCamelCase       ,  """vocab.txt"""							)
               with open(_lowerCamelCase       ,  """w"""       ,  encoding="""utf-8"""							) as vocab_writer:
                     vocab_writer.write("""""".join([x + """\n""" for x in self.vocab_tokens]							)							)
               __lowerCamelCase     :      List[str]       						=						CustomTokenizer(_lowerCamelCase							)

         # No fast custom tokenizer
         tokenizer.push_to_hub("""test-dynamic-tokenizer"""       ,  use_auth_token=self._token							)

         __lowerCamelCase     :      Any       						=						AutoTokenizer.from_pretrained(F"""{USER}/test-dynamic-tokenizer"""       ,  trust_remote_code=_lowerCamelCase							)
         # Can't make an isinstance check because the new_model.config is from the CustomTokenizer class of a dynamic module
         self.assertEqual(tokenizer.__class__.__name__       ,  """CustomTokenizer"""							)

         # Fast and slow custom tokenizer
         CustomTokenizerFast.register_for_auto_class()
         with tempfile.TemporaryDirectory() as tmp_dir:
               __lowerCamelCase     :      int       						=						os.path.join(_lowerCamelCase       ,  """vocab.txt"""							)
               with open(_lowerCamelCase       ,  """w"""       ,  encoding="""utf-8"""							) as vocab_writer:
                     vocab_writer.write("""""".join([x + """\n""" for x in self.vocab_tokens]							)							)

               __lowerCamelCase     :      Union[str, Any]       						=						BertTokenizerFast.from_pretrained(_lowerCamelCase							)
               bert_tokenizer.save_pretrained(_lowerCamelCase							)
               __lowerCamelCase     :      Optional[Any]       						=						CustomTokenizerFast.from_pretrained(_lowerCamelCase							)

         tokenizer.push_to_hub("""test-dynamic-tokenizer"""       ,  use_auth_token=self._token							)

         __lowerCamelCase     :      List[Any]       						=						AutoTokenizer.from_pretrained(F"""{USER}/test-dynamic-tokenizer"""       ,  trust_remote_code=_lowerCamelCase							)
         # Can't make an isinstance check because the new_model.config is from the FakeConfig class of a dynamic module
         self.assertEqual(tokenizer.__class__.__name__       ,  """CustomTokenizerFast"""							)
         __lowerCamelCase     :      Dict       						=						AutoTokenizer.from_pretrained(
             F"""{USER}/test-dynamic-tokenizer"""       ,  use_fast=_lowerCamelCase       ,  trust_remote_code=_lowerCamelCase							)
         # Can't make an isinstance check because the new_model.config is from the FakeConfig class of a dynamic module
         self.assertEqual(tokenizer.__class__.__name__       ,  """CustomTokenizer"""							)







class 							_UpperCamelCase (						unittest.TestCase						):

   '''simple docstring'''







   def   _snake_case    (  self  : Optional[int]							):






         '''simple docstring'''



         __lowerCamelCase     :      List[str]       						=						Trie()
         trie.add("""Hello 友達"""							)
         self.assertEqual(trie.data       ,  {"""H""": {"""e""": {"""l""": {"""l""": {"""o""": {""" """: {"""友""": {"""達""": {"""""": 1}}}}}}}}}							)
         trie.add("""Hello"""							)
         trie.data
         self.assertEqual(trie.data       ,  {"""H""": {"""e""": {"""l""": {"""l""": {"""o""": {"""""": 1, """ """: {"""友""": {"""達""": {"""""": 1}}}}}}}}}							)

   def   _snake_case    (  self  : Tuple							):






         '''simple docstring'''



         __lowerCamelCase     :      int       						=						Trie()
         self.assertEqual(trie.split("""[CLS] This is a extra_id_100"""							)       ,  ["""[CLS] This is a extra_id_100"""]							)
         trie.add("""[CLS]"""							)
         trie.add("""extra_id_1"""							)
         trie.add("""extra_id_100"""							)
         self.assertEqual(trie.split("""[CLS] This is a extra_id_100"""							)       ,  ["""[CLS]""", """ This is a """, """extra_id_100"""]							)

   def   _snake_case    (  self  : Optional[int]							):






         '''simple docstring'''



         __lowerCamelCase     :      str       						=						Trie()
         trie.add("""A"""							)
         self.assertEqual(trie.split("""ABC"""							)       ,  ["""A""", """BC"""]							)
         self.assertEqual(trie.split("""BCA"""							)       ,  ["""BC""", """A"""]							)

   def   _snake_case    (  self  : Optional[int]							):






         '''simple docstring'''



         __lowerCamelCase     :      Tuple       						=						Trie()
         trie.add("""TOKEN]"""							)
         trie.add("""[SPECIAL_TOKEN]"""							)
         self.assertEqual(trie.split("""This is something [SPECIAL_TOKEN]"""							)       ,  ["""This is something """, """[SPECIAL_TOKEN]"""]							)

   def   _snake_case    (  self  : Dict							):






         '''simple docstring'''



         __lowerCamelCase     :      Union[str, Any]       						=						Trie()
         trie.add("""A"""							)
         trie.add("""P"""							)
         trie.add("""[SPECIAL_TOKEN]"""							)
         self.assertEqual(trie.split("""This is something [SPECIAL_TOKEN]"""							)       ,  ["""This is something """, """[SPECIAL_TOKEN]"""]							)

   def   _snake_case    (  self  : Tuple							):






         '''simple docstring'''



         __lowerCamelCase     :      List[Any]       						=						Trie()
         trie.add("""AB"""							)
         trie.add("""B"""							)
         trie.add("""C"""							)
         self.assertEqual(trie.split("""ABC"""							)       ,  ["""AB""", """C"""]							)

   def   _snake_case    (  self  : List[str]							):






         '''simple docstring'''



         __lowerCamelCase     :      List[str]       						=						Trie()
         trie.add("""ABC"""							)
         trie.add("""B"""							)
         trie.add("""CD"""							)
         self.assertEqual(trie.split("""ABCD"""							)       ,  ["""ABC""", """D"""]							)


   def   _snake_case    (  self  : int							):






         '''simple docstring'''



         __lowerCamelCase     :      Optional[Any]       						=						Trie()
         __lowerCamelCase     :      Optional[int]       						=						trie.cut_text("""ABC"""       ,  [0, 0, 2, 1, 2, 3]							)
         self.assertEqual(_lowerCamelCase       ,  ["""AB""", """C"""]							)







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



