

"""simple docstring"""






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        _lowercase					(     __UpperCAmelCase   , unittest.TestCase     ):
     _lowerCamelCase     =       '''ssube/stable-diffusion-x4-upscaler-onnx'''







     def 	lowerCAmelCase__    (      self ,						UpperCamelCase_=0       ):
      __magic_name__     =      floats_tensor((1, 3, 128, 128) ,						rng=random.Random(lowercase_       )       )
      __magic_name__     =      torch.manual_seed(lowercase_       )
      __magic_name__     =      {
          '''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 	lowerCAmelCase__    (      self       ):
      __magic_name__     =      OnnxStableDiffusionUpscalePipeline.from_pretrained(self.hub_checkpoint ,						provider='''CPUExecutionProvider'''       )
      pipe.set_progress_bar_config(disable=lowercase_       )

      __magic_name__     =      self.get_dummy_inputs()
      __magic_name__     =      pipe(**lowercase_       ).images
      __magic_name__     =      image[0, -3:, -3:, -1].flatten()

      # started as 128, should now be 512
      assert image.shape == (1, 512, 512, 3)
      __magic_name__     =      np.array(
          [0.6_9_7_4_7_8_2, 0.6_8_9_0_2_0_9_3, 0.7_0_1_3_5_8_8_5, 0.7_5_8_3_6_1_8, 0.7_8_0_4_5_4_5, 0.7_8_5_4_9_1_2, 0.7_8_6_6_7_4_2_6, 0.7_8_7_4_3_8_6_3, 0.7_8_0_7_0_2_2_3]       )
      assert np.abs(image_slice - expected_slice       ).max() < 1E-1







     def 	lowerCAmelCase__    (      self       ):
      __magic_name__     =      OnnxStableDiffusionUpscalePipeline.from_pretrained(self.hub_checkpoint ,						provider='''CPUExecutionProvider'''       )
      __magic_name__     =      PNDMScheduler.from_config(pipe.scheduler.config ,						skip_prk_steps=lowercase_       )
      pipe.set_progress_bar_config(disable=lowercase_       )

      __magic_name__     =      self.get_dummy_inputs()
      __magic_name__     =      pipe(**lowercase_       ).images
      __magic_name__     =      image[0, -3:, -3:, -1]

      assert image.shape == (1, 512, 512, 3)
      __magic_name__     =      np.array(
          [0.6_8_9_8_8_9_2, 0.5_9_2_4_0_5_5_6, 0.5_2_4_9_9_5_2_7, 0.5_8_8_6_6_2_1_5, 0.5_2_2_5_8_2_3_5, 0.5_2_5_7_2_7_1_5, 0.6_2_4_1_4_4_7_3, 0.6_1_7_4_3_8_7, 0.6_2_1_4_9_6_4]       )
      assert np.abs(image_slice.flatten() - expected_slice       ).max() < 1E-1







     def 	lowerCAmelCase__    (      self       ):
      __magic_name__     =      OnnxStableDiffusionUpscalePipeline.from_pretrained(self.hub_checkpoint ,						provider='''CPUExecutionProvider'''       )
      __magic_name__     =      DPMSolverMultistepScheduler.from_config(pipe.scheduler.config       )
      pipe.set_progress_bar_config(disable=lowercase_       )

      __magic_name__     =      self.get_dummy_inputs()
      __magic_name__     =      pipe(**lowercase_       ).images
      __magic_name__     =      image[0, -3:, -3:, -1]

      assert image.shape == (1, 512, 512, 3)
      __magic_name__     =      np.array(
          [0.7_6_5_9_2_7_8, 0.7_6_4_3_7_6_6_4, 0.7_5_5_7_9_1_0_7, 0.7_6_9_1_1_1_6, 0.7_7_6_6_6_9_8_6, 0.7_7_2_7_6_7_2, 0.7_7_5_8_6_6_4, 0.7_8_1_2_2_2_6, 0.7_6_9_4_2_5_1_5]       )

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







     def 	lowerCAmelCase__    (      self       ):
      __magic_name__     =      OnnxStableDiffusionUpscalePipeline.from_pretrained(self.hub_checkpoint ,						provider='''CPUExecutionProvider'''       )
      __magic_name__     =      EulerDiscreteScheduler.from_config(pipe.scheduler.config       )
      pipe.set_progress_bar_config(disable=lowercase_       )

      __magic_name__     =      self.get_dummy_inputs()
      __magic_name__     =      pipe(**lowercase_       ).images
      __magic_name__     =      image[0, -3:, -3:, -1]

      assert image.shape == (1, 512, 512, 3)
      __magic_name__     =      np.array(
          [0.6_9_7_4_7_8_2, 0.6_8_9_0_2_0_9_3, 0.7_0_1_3_5_8_8_5, 0.7_5_8_3_6_1_8, 0.7_8_0_4_5_4_5, 0.7_8_5_4_9_1_2, 0.7_8_6_6_7_4_2_6, 0.7_8_7_4_3_8_6_3, 0.7_8_0_7_0_2_2_3]       )
      assert np.abs(image_slice.flatten() - expected_slice       ).max() < 1E-1







     def 	lowerCAmelCase__    (      self       ):
      __magic_name__     =      OnnxStableDiffusionUpscalePipeline.from_pretrained(self.hub_checkpoint ,						provider='''CPUExecutionProvider'''       )
      __magic_name__     =      EulerAncestralDiscreteScheduler.from_config(pipe.scheduler.config       )
      pipe.set_progress_bar_config(disable=lowercase_       )

      __magic_name__     =      self.get_dummy_inputs()
      __magic_name__     =      pipe(**lowercase_       ).images
      __magic_name__     =      image[0, -3:, -3:, -1]

      assert image.shape == (1, 512, 512, 3)
      __magic_name__     =      np.array(
          [0.7_7_4_2_4_4_9_6, 0.7_7_3_6_0_1, 0.7_6_4_5_2_8_8, 0.7_7_6_9_5_9_8, 0.7_7_7_2_7_3_9, 0.7_7_3_8_6_8_8, 0.7_8_1_8_7_2_3_3, 0.7_7_8_7_9_5_8_4, 0.7_6_7_0_4_3]       )

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


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







     @property
     def 	lowerCAmelCase__    (      self       ):
      return (
          "CUDAExecutionProvider",
          {
              "gpu_mem_limit": "15000000000",  # 15GB
              "arena_extend_strategy": "kSameAsRequested",
          },
      )







     @property
     def 	lowerCAmelCase__    (      self       ):
      __magic_name__     =      ort.SessionOptions()
      __magic_name__     =      False
      return options







     def 	lowerCAmelCase__    (      self       ):
      __magic_name__     =      load_image(
          '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
          '''/img2img/sketch-mountains-input.jpg'''       )
      __magic_name__     =      init_image.resize((128, 128)       )
      # using the PNDM scheduler by default
      __magic_name__     =      OnnxStableDiffusionUpscalePipeline.from_pretrained(
          '''ssube/stable-diffusion-x4-upscaler-onnx''' ,						provider=self.gpu_provider ,						sess_options=self.gpu_options ,						)
      pipe.set_progress_bar_config(disable=lowercase_       )

      __magic_name__     =      '''A fantasy landscape, trending on artstation'''

      __magic_name__     =      torch.manual_seed(0       )
      __magic_name__     =      pipe(
          prompt=lowercase_ ,						image=lowercase_ ,						guidance_scale=7.5 ,						num_inference_steps=10 ,						generator=lowercase_ ,						output_type='''np''' ,						)
      __magic_name__     =      output.images
      __magic_name__     =      images[0, 255:258, 383:386, -1]

      assert images.shape == (1, 512, 512, 3)
      __magic_name__     =      np.array([0.4_8_8_3, 0.4_9_4_7, 0.4_9_8_0, 0.4_9_7_5, 0.4_9_8_2, 0.4_9_8_0, 0.5_0_0_0, 0.5_0_0_6, 0.4_9_7_2]       )
      # TODO: lower the tolerance after finding the cause of onnxruntime reproducibility issues

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







     def 	lowerCAmelCase__    (      self       ):
      __magic_name__     =      load_image(
          '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
          '''/img2img/sketch-mountains-input.jpg'''       )
      __magic_name__     =      init_image.resize((128, 128)       )
      __magic_name__     =      LMSDiscreteScheduler.from_pretrained(
          '''ssube/stable-diffusion-x4-upscaler-onnx''' ,						subfolder='''scheduler'''       )
      __magic_name__     =      OnnxStableDiffusionUpscalePipeline.from_pretrained(
          '''ssube/stable-diffusion-x4-upscaler-onnx''' ,						scheduler=lowercase_ ,						provider=self.gpu_provider ,						sess_options=self.gpu_options ,						)
      pipe.set_progress_bar_config(disable=lowercase_       )

      __magic_name__     =      '''A fantasy landscape, trending on artstation'''

      __magic_name__     =      torch.manual_seed(0       )
      __magic_name__     =      pipe(
          prompt=lowercase_ ,						image=lowercase_ ,						guidance_scale=7.5 ,						num_inference_steps=20 ,						generator=lowercase_ ,						output_type='''np''' ,						)
      __magic_name__     =      output.images
      __magic_name__     =      images[0, 255:258, 383:386, -1]

      assert images.shape == (1, 512, 512, 3)
      __magic_name__     =      np.array(
          [0.5_0_1_7_3_7_5_3, 0.5_0_2_2_3_3_5_6, 0.5_0_2_0_3_9, 0.5_0_2_3_3_0_3_6, 0.5_0_2_3_7_2_5, 0.5_0_2_2_6_0_1, 0.5_0_1_8_7_5_8, 0.5_0_2_3_4_0_8_5, 0.5_0_2_4_1_5_6_6]       )
      # TODO: lower the tolerance after finding the cause of onnxruntime reproducibility issues

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




"""simple docstring"""






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

from requests.exceptions import HTTPError

from transformers.utils import (
    CONFIG_NAME,
    FLAX_WEIGHTS_NAME,
    TF2_WEIGHTS_NAME,
    TRANSFORMERS_CACHE,
    WEIGHTS_NAME,
    cached_file,
    get_file_from_repo,
    has_file,
)


__lowerCamelCase										=		"hf-internal-testing/tiny-random-bert"
__lowerCamelCase										=		os.path.join(TRANSFORMERS_CACHE, "models--hf-internal-testing--tiny-random-bert")
__lowerCamelCase										=		"9b8c223d42b2188cb49d29af482996f9d0f3e5a6"
class        _lowercase					(     unittest.TestCase     ):







     def 	lowerCAmelCase__    (      self       ):
      __magic_name__     =      cached_file(UpperCamelCase_ ,						UpperCamelCase_       )
      # Should have downloaded the file in here
      self.assertTrue(os.path.isdir(UpperCamelCase_       )       )
      # Cache should contain at least those three subfolders:
      for subfolder in ["blobs", "refs", "snapshots"]:
       self.assertTrue(os.path.isdir(os.path.join(UpperCamelCase_ ,						UpperCamelCase_       )       )       )
      with open(os.path.join(UpperCamelCase_ ,						'''refs''' ,						'''main'''       )       ) as f:
       __magic_name__     =      f.read()
      self.assertEqual(UpperCamelCase_ ,						os.path.join(UpperCamelCase_ ,						'''snapshots''' ,						UpperCamelCase_ ,						UpperCamelCase_       )       )
      self.assertTrue(os.path.isfile(UpperCamelCase_       )       )

      # File is cached at the same place the second time.
      __magic_name__     =      cached_file(UpperCamelCase_ ,						UpperCamelCase_       )
      self.assertEqual(UpperCamelCase_ ,						UpperCamelCase_       )

      # Using a specific revision to test the full commit hash.
      __magic_name__     =      cached_file(UpperCamelCase_ ,						UpperCamelCase_ ,						revision='''9b8c223'''       )
      self.assertEqual(UpperCamelCase_ ,						os.path.join(UpperCamelCase_ ,						'''snapshots''' ,						UpperCamelCase_ ,						UpperCamelCase_       )       )







     def 	lowerCAmelCase__    (      self       ):
      with self.assertRaisesRegex(UpperCamelCase_ ,						'''is not a valid model identifier'''       ):
       __magic_name__     =      cached_file('''tiny-random-bert''' ,						UpperCamelCase_       )

      with self.assertRaisesRegex(UpperCamelCase_ ,						'''is not a valid git identifier'''       ):
       __magic_name__     =      cached_file(UpperCamelCase_ ,						UpperCamelCase_ ,						revision='''aaaa'''       )

      with self.assertRaisesRegex(UpperCamelCase_ ,						'''does not appear to have a file named'''       ):
       __magic_name__     =      cached_file(UpperCamelCase_ ,						'''conf'''       )







     def 	lowerCAmelCase__    (      self       ):
      with self.assertRaisesRegex(UpperCamelCase_ ,						'''does not appear to have a file named'''       ):
       __magic_name__     =      cached_file(UpperCamelCase_ ,						'''conf'''       )

      with open(os.path.join(UpperCamelCase_ ,						'''refs''' ,						'''main'''       )       ) as f:
       __magic_name__     =      f.read()
      self.assertTrue(os.path.isfile(os.path.join(UpperCamelCase_ ,						'''.no_exist''' ,						UpperCamelCase_ ,						'''conf'''       )       )       )

      __magic_name__     =      cached_file(UpperCamelCase_ ,						'''conf''' ,						_raise_exceptions_for_missing_entries=UpperCamelCase_       )
      self.assertIsNone(UpperCamelCase_       )

      __magic_name__     =      cached_file(UpperCamelCase_ ,						'''conf''' ,						local_files_only=UpperCamelCase_ ,						_raise_exceptions_for_missing_entries=UpperCamelCase_       )
      self.assertIsNone(UpperCamelCase_       )

      __magic_name__     =      mock.Mock()
      __magic_name__     =      500
      __magic_name__     =      {}
      __magic_name__     =      HTTPError
      __magic_name__     =      {}

      # Under the mock environment we get a 500 error when trying to reach the tokenizer.
      with mock.patch('''requests.Session.request''' ,						return_value=UpperCamelCase_       ) as mock_head:
       __magic_name__     =      cached_file(UpperCamelCase_ ,						'''conf''' ,						_raise_exceptions_for_connection_errors=UpperCamelCase_       )
       self.assertIsNone(UpperCamelCase_       )
       # This check we did call the fake head request
       mock_head.assert_called()







     def 	lowerCAmelCase__    (      self       ):
      self.assertTrue(has_file('''hf-internal-testing/tiny-bert-pt-only''' ,						UpperCamelCase_       )       )
      self.assertFalse(has_file('''hf-internal-testing/tiny-bert-pt-only''' ,						UpperCamelCase_       )       )
      self.assertFalse(has_file('''hf-internal-testing/tiny-bert-pt-only''' ,						UpperCamelCase_       )       )







     def 	lowerCAmelCase__    (      self       ):
      # `get_file_from_repo` returns None if the file does not exist
      self.assertIsNone(get_file_from_repo('''bert-base-cased''' ,						'''ahah.txt'''       )       )

      # The function raises if the repository does not exist.
      with self.assertRaisesRegex(UpperCamelCase_ ,						'''is not a valid model identifier'''       ):
       get_file_from_repo('''bert-base-case''' ,						UpperCamelCase_       )

      # The function raises if the revision does not exist.
      with self.assertRaisesRegex(UpperCamelCase_ ,						'''is not a valid git identifier'''       ):
       get_file_from_repo('''bert-base-cased''' ,						UpperCamelCase_ ,						revision='''ahaha'''       )

      __magic_name__     =      get_file_from_repo('''bert-base-cased''' ,						UpperCamelCase_       )
      # The name is the cached name which is not very easy to test, so instead we load the content.
      __magic_name__     =      json.loads(open(UpperCamelCase_ ,						'''r'''       ).read()       )
      self.assertEqual(config['''hidden_size'''] ,						768       )







     def 	lowerCAmelCase__    (      self       ):
      with tempfile.TemporaryDirectory() as tmp_dir:
       __magic_name__     =      Path(UpperCamelCase_       ) / '''a.txt'''
       filename.touch()
       self.assertEqual(get_file_from_repo(UpperCamelCase_ ,						'''a.txt'''       ) ,						str(UpperCamelCase_       )       )

       self.assertIsNone(get_file_from_repo(UpperCamelCase_ ,						'''b.txt'''       )       )

