

"""simple docstring"""






import random
import unittest

import numpy as np

from diffusers import (
    DPMSolverMultistepScheduler,
    EulerAncestralDiscreteScheduler,
    EulerDiscreteScheduler,
    LMSDiscreteScheduler,
    OnnxStableDiffusionImgaImgPipeline,
    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     =       '''hf-internal-testing/tiny-random-OnnxStableDiffusionPipeline'''







     def 	lowerCAmelCase__    (      self ,						UpperCamelCase_=0       ):
      __magic_name__     =      floats_tensor((1, 3, 128, 128) ,						rng=random.Random(__A       )       )
      __magic_name__     =      np.random.RandomState(__A       )
      __magic_name__     =      {
          "prompt": "A painting of a squirrel eating a burger",
          "image": image,
          "generator": generator,
          "num_inference_steps": 3,
          "strength": 0.7_5,
          "guidance_scale": 7.5,
          "output_type": "numpy",
      }
      return inputs







     def 	lowerCAmelCase__    (      self       ):
      __magic_name__     =      OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint ,						provider='''CPUExecutionProvider'''       )
      pipe.set_progress_bar_config(disable=__A       )

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

      assert image.shape == (1, 128, 128, 3)
      __magic_name__     =      np.array([0.6_9_6_4_3, 0.5_8_4_8_4, 0.5_0_3_1_4, 0.5_8_7_6_0, 0.5_5_3_6_8, 0.5_9_6_4_3, 0.5_1_5_2_9, 0.4_1_2_1_7, 0.4_9_0_8_7]       )
      assert np.abs(image_slice - expected_slice       ).max() < 1E-1







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

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

      assert image.shape == (1, 128, 128, 3)
      __magic_name__     =      np.array([0.6_1_7_3_7, 0.5_4_6_4_2, 0.5_3_1_8_3, 0.5_4_4_6_5, 0.5_2_7_4_2, 0.6_0_5_2_5, 0.4_9_9_6_9, 0.4_0_6_5_5, 0.4_8_1_5_4]       )

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







     def 	lowerCAmelCase__    (      self       ):
      __magic_name__     =      OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint ,						provider='''CPUExecutionProvider'''       )
      __magic_name__     =      LMSDiscreteScheduler.from_config(pipe.scheduler.config       )
      pipe.set_progress_bar_config(disable=__A       )

      # warmup pass to apply optimizations
      __magic_name__     =      pipe(**self.get_dummy_inputs()       )

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

      assert image.shape == (1, 128, 128, 3)
      __magic_name__     =      np.array([0.5_2_7_6_1, 0.5_9_9_7_7, 0.4_9_0_3_3, 0.4_9_6_1_9, 0.5_4_2_8_2, 0.5_0_3_1_1, 0.4_7_6_0_0, 0.4_0_9_1_8, 0.4_5_2_0_3]       )

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







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

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

      assert image.shape == (1, 128, 128, 3)
      __magic_name__     =      np.array([0.5_2_9_1_1, 0.6_0_0_0_4, 0.4_9_2_2_9, 0.4_9_8_0_5, 0.5_4_5_0_2, 0.5_0_6_8_0, 0.4_7_7_7_7, 0.4_1_0_2_8, 0.4_5_3_0_4]       )

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







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

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

      assert image.shape == (1, 128, 128, 3)
      __magic_name__     =      np.array([0.5_2_9_1_1, 0.6_0_0_0_4, 0.4_9_2_2_9, 0.4_9_8_0_5, 0.5_4_5_0_2, 0.5_0_6_8_0, 0.4_7_7_7_7, 0.4_1_0_2_8, 0.4_5_3_0_4]       )

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







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

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

      assert image.shape == (1, 128, 128, 3)
      __magic_name__     =      np.array([0.6_5_3_3_1, 0.5_8_2_7_7, 0.4_8_2_0_4, 0.5_6_0_5_9, 0.5_3_6_6_5, 0.5_6_2_3_5, 0.5_0_9_6_9, 0.4_0_0_0_9, 0.4_6_5_5_2]       )

      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((768, 512)       )
      # using the PNDM scheduler by default
      __magic_name__     =      OnnxStableDiffusionImgaImgPipeline.from_pretrained(
          '''CompVis/stable-diffusion-v1-4''' ,						revision='''onnx''' ,						safety_checker=__A ,						feature_extractor=__A ,						provider=self.gpu_provider ,						sess_options=self.gpu_options ,						)
      pipe.set_progress_bar_config(disable=__A       )

      __magic_name__     =      "A fantasy landscape, trending on artstation"

      __magic_name__     =      np.random.RandomState(0       )
      __magic_name__     =      pipe(
          prompt=__A ,						image=__A ,						strength=0.7_5 ,						guidance_scale=7.5 ,						num_inference_steps=10 ,						generator=__A ,						output_type='''np''' ,						)
      __magic_name__     =      output.images
      __magic_name__     =      images[0, 255:258, 383:386, -1]

      assert images.shape == (1, 512, 768, 3)
      __magic_name__     =      np.array([0.4_9_0_9, 0.5_0_5_9, 0.5_3_7_2, 0.4_6_2_3, 0.4_8_7_6, 0.5_0_4_9, 0.4_8_2_0, 0.4_9_5_6, 0.5_0_1_9]       )
      # 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((768, 512)       )
      __magic_name__     =      LMSDiscreteScheduler.from_pretrained(
          '''runwayml/stable-diffusion-v1-5''' ,						subfolder='''scheduler''' ,						revision='''onnx'''       )
      __magic_name__     =      OnnxStableDiffusionImgaImgPipeline.from_pretrained(
          '''runwayml/stable-diffusion-v1-5''' ,						revision='''onnx''' ,						scheduler=__A ,						safety_checker=__A ,						feature_extractor=__A ,						provider=self.gpu_provider ,						sess_options=self.gpu_options ,						)
      pipe.set_progress_bar_config(disable=__A       )

      __magic_name__     =      "A fantasy landscape, trending on artstation"

      __magic_name__     =      np.random.RandomState(0       )
      __magic_name__     =      pipe(
          prompt=__A ,						image=__A ,						strength=0.7_5 ,						guidance_scale=7.5 ,						num_inference_steps=20 ,						generator=__A ,						output_type='''np''' ,						)
      __magic_name__     =      output.images
      __magic_name__     =      images[0, 255:258, 383:386, -1]

      assert images.shape == (1, 512, 768, 3)
      __magic_name__     =      np.array([0.8_0_4_3, 0.9_2_6, 0.9_5_8_1, 0.8_1_1_9, 0.8_9_5_4, 0.9_1_3, 0.7_2_0_9, 0.7_4_6_3, 0.7_4_3_1]       )
      # 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 html

from ...feature_extraction_utils import BatchFeature, FeatureExtractionMixin
from ...utils import is_bsa_available, logging, requires_backends


if is_bsa_available():
 import bsa
 from bsa import BeautifulSoup


__lowerCamelCase										=		logging.get_logger(__name__)
class        _lowercase					(     __UpperCAmelCase     ):







     def __init__(      self ,						**UpperCamelCase_       ):
      requires_backends(self ,						['''bs4''']       )
      super().__init__(**UpperCamelCase_       )







     def 	lowerCAmelCase__    (      self ,						UpperCamelCase_       ):
      __magic_name__     =      []
      __magic_name__     =      []
      __magic_name__     =      element if element.name else element.parent
      for parent in child.parents:  # type: bs4.element.Tag
       __magic_name__     =      parent.find_all(child.name ,						recursive=UpperCamelCase_       )
       xpath_tags.append(child.name       )
       xpath_subscripts.append(
           0 if 1 == len(UpperCamelCase_       ) else next(i for i, s in enumerate(UpperCamelCase_ ,						1       ) if s is child       )       )
       __magic_name__     =      parent
      xpath_tags.reverse()
      xpath_subscripts.reverse()
      return xpath_tags, xpath_subscripts







     def 	lowerCAmelCase__    (      self ,						UpperCamelCase_       ):
      __magic_name__     =      BeautifulSoup(UpperCamelCase_ ,						'''html.parser'''       )

      __magic_name__     =      []
      __magic_name__     =      []
      __magic_name__     =      []

      for element in html_code.descendants:
       if type(UpperCamelCase_       ) == bsa.element.NavigableString:
        if type(element.parent       ) != bsa.element.Tag:
         continue

        __magic_name__     =      html.unescape(UpperCamelCase_       ).strip()
        if not text_in_this_tag:
         continue

        all_doc_strings.append(UpperCamelCase_       )

        __magic_name__   ,						__magic_name__     =      self.xpath_soup(UpperCamelCase_       )
        stringaxtag_seq.append(UpperCamelCase_       )
        stringaxsubs_seq.append(UpperCamelCase_       )

      if len(UpperCamelCase_       ) != len(UpperCamelCase_       ):
       raise ValueError('''Number of doc strings and xtags does not correspond'''       )
      if len(UpperCamelCase_       ) != len(UpperCamelCase_       ):
       raise ValueError('''Number of doc strings and xsubs does not correspond'''       )

      return all_doc_strings, stringaxtag_seq, stringaxsubs_seq







     def 	lowerCAmelCase__    (      self ,						UpperCamelCase_ ,						UpperCamelCase_       ):
      __magic_name__     =      ''''''
      for tagname, subs in zip(UpperCamelCase_ ,						UpperCamelCase_       ):
       xpath += f'''/{tagname}'''
       if subs != 0:
        xpath += f'''[{subs}]'''
      return xpath







     def __call__(      self ,						UpperCamelCase_       ):
      __magic_name__     =      False

      # Check that strings has a valid type
      if isinstance(UpperCamelCase_ ,						UpperCamelCase_       ):
       __magic_name__     =      True
      elif isinstance(UpperCamelCase_ ,						(list, tuple)       ):
       if len(UpperCamelCase_       ) == 0 or isinstance(html_strings[0] ,						UpperCamelCase_       ):
        __magic_name__     =      True

      if not valid_strings:
       raise ValueError(
           '''HTML strings must of type `str`, `List[str]` (batch of examples), '''
           f'''but is of type {type(UpperCamelCase_       )}.'''       )

      __magic_name__     =      bool(isinstance(UpperCamelCase_ ,						(list, tuple)       ) and (isinstance(html_strings[0] ,						UpperCamelCase_       ))       )

      if not is_batched:
       __magic_name__     =      [html_strings]

      # Get nodes + xpaths
      __magic_name__     =      []
      __magic_name__     =      []
      for html_string in html_strings:
       __magic_name__   ,						__magic_name__   ,						__magic_name__     =      self.get_three_from_single(UpperCamelCase_       )
       nodes.append(UpperCamelCase_       )
       __magic_name__     =      []
       for node, tag_list, sub_list in zip(UpperCamelCase_ ,						UpperCamelCase_ ,						UpperCamelCase_       ):
        __magic_name__     =      self.construct_xpath(UpperCamelCase_ ,						UpperCamelCase_       )
        xpath_strings.append(UpperCamelCase_       )
       xpaths.append(UpperCamelCase_       )

      # return as Dict
      __magic_name__     =      {'''nodes''': nodes, '''xpaths''': xpaths}
      __magic_name__     =      BatchFeature(data=UpperCamelCase_ ,						tensor_type=UpperCamelCase_       )

      return encoded_inputs

