



'''simple docstring'''





import argparse
import re

import numpy as np
import requests
import torch
from huggingface_hub import hf_hub_download
from PIL import Image

from transformers import (
    SamConfig,
    SamImageProcessor,
    SamModel,
    SamProcessor,
    SamVisionConfig,
)


lowercase_	 =						{
    "iou_prediction_head.layers.0": "iou_prediction_head.proj_in",
    "iou_prediction_head.layers.1": "iou_prediction_head.layers.0",
    "iou_prediction_head.layers.2": "iou_prediction_head.proj_out",
    "mask_decoder.output_upscaling.0": "mask_decoder.upscale_conv1",
    "mask_decoder.output_upscaling.1": "mask_decoder.upscale_layer_norm",
    "mask_decoder.output_upscaling.3": "mask_decoder.upscale_conv2",
    "mask_downscaling.0": "mask_embed.conv1",
    "mask_downscaling.1": "mask_embed.layer_norm1",
    "mask_downscaling.3": "mask_embed.conv2",
    "mask_downscaling.4": "mask_embed.layer_norm2",
    "mask_downscaling.6": "mask_embed.conv3",
    "point_embeddings": "point_embed",
    "pe_layer.positional_encoding_gaussian_matrix": "shared_embedding.positional_embedding",
    "image_encoder": "vision_encoder",
    "neck.0": "neck.conv1",
    "neck.1": "neck.layer_norm1",
    "neck.2": "neck.conv2",
    "neck.3": "neck.layer_norm2",
    "patch_embed.proj": "patch_embed.projection",
    ".norm": ".layer_norm",
    "blocks": "layers",
}




def lowerCAmelCase						(__A):



     """simple docstring"""



     _a											=						{}
     state_dict.pop('''pixel_mean'''						,			snake_case_)
     state_dict.pop('''pixel_std'''						,			snake_case_)

     _a											=						r""".*.output_hypernetworks_mlps.(\d+).layers.(\d+).*"""

     for key, value in state_dict.items():
          for key_to_modify, new_key in KEYS_TO_MODIFY_MAPPING.items():
               if key_to_modify in key:
                    _a											=						key.replace(snake_case_						,			snake_case_)

          if re.match(snake_case_						,			snake_case_):
               _a											=						int(re.match(snake_case_						,			snake_case_).group(2))
               if layer_nb == 0:
                    _a											=						key.replace('''layers.0'''						,			'''proj_in''')
               elif layer_nb == 1:
                    _a											=						key.replace('''layers.1'''						,			'''layers.0''')
               elif layer_nb == 2:
                    _a											=						key.replace('''layers.2'''						,			'''proj_out''')

          _a											=						value

     _a											=						model_state_dict[
         """prompt_encoder.shared_embedding.positional_embedding"""
     ]

     return model_state_dict




def lowerCAmelCase						(__A						,			__A						,			__A						,			__A="ybelkada/segment-anything"):



     """simple docstring"""



     _a											=						hf_hub_download(snake_case_						,			F'''checkpoints/{model_name}.pth''')

     if "sam_vit_b" in model_name:
          _a											=						SamConfig()
     elif "sam_vit_l" in model_name:
          _a											=						SamVisionConfig(
              hidden_size=1_024						,			num_hidden_layers=24						,			num_attention_heads=16						,			global_attn_indexes=[5, 11, 17, 23]						,			)

          _a											=						SamConfig(
              vision_config=snake_case_						,			)
     elif "sam_vit_h" in model_name:
          _a											=						SamVisionConfig(
              hidden_size=1_280						,			num_hidden_layers=32						,			num_attention_heads=16						,			global_attn_indexes=[7, 15, 23, 31]						,			)

          _a											=						SamConfig(
              vision_config=snake_case_						,			)

     _a											=						torch.load(snake_case_						,			map_location='''cpu''')
     _a											=						replace_keys(snake_case_)

     _a											=						SamImageProcessor()

     _a											=						SamProcessor(image_processor=snake_case_)
     _a											=						SamModel(snake_case_)

     hf_model.load_state_dict(snake_case_)
     _a											=						hf_model.to('''cuda''')

     _a											=						"""https://huggingface.co/ybelkada/segment-anything/resolve/main/assets/car.png"""
     _a											=						Image.open(requests.get(snake_case_						,			stream=snake_case_).raw).convert('''RGB''')

     _a											=						[[[400, 650]]]
     _a											=						[[1]]

     _a											=						processor(images=np.array(snake_case_)						,			return_tensors='''pt''').to('''cuda''')

     with torch.no_grad():
          _a											=						hf_model(**snake_case_)
     _a											=						output.iou_scores.squeeze()

     if model_name == "sam_vit_h_4b8939":
          assert scores[-1].item() == 0.5_79_89_02_51_15_96_68

          _a											=						processor(
              images=np.array(snake_case_)						,			input_points=snake_case_						,			input_labels=snake_case_						,			return_tensors='''pt''').to('''cuda''')

          with torch.no_grad():
               _a											=						hf_model(**snake_case_)
          _a											=						output.iou_scores.squeeze()

          assert scores[-1].item() == 0.97_12_60_30_92_19_36_04

          _a											=						((75, 275, 1_725, 850),)

          _a											=						processor(images=np.array(snake_case_)						,			input_boxes=snake_case_						,			return_tensors='''pt''').to('''cuda''')

          with torch.no_grad():
               _a											=						hf_model(**snake_case_)
          _a											=						output.iou_scores.squeeze()

          assert scores[-1].item() == 0.86_86_01_56_05_92_65_14

          # Test with 2 points and 1 image.
          _a											=						[[[400, 650], [800, 650]]]
          _a											=						[[1, 1]]

          _a											=						processor(
              images=np.array(snake_case_)						,			input_points=snake_case_						,			input_labels=snake_case_						,			return_tensors='''pt''').to('''cuda''')

          with torch.no_grad():
               _a											=						hf_model(**snake_case_)
          _a											=						output.iou_scores.squeeze()

          assert scores[-1].item() == 0.99_36_04_77_92_43_46_92


if __name__ == "__main__":
  lowercase_	 =						argparse.ArgumentParser()
  lowercase_	 =						["sam_vit_b_01ec64", "sam_vit_h_4b8939", "sam_vit_l_0b3195"]
  parser.add_argument(
      "--model_name",
      default="sam_vit_h_4b8939",
      choices=choices,
      type=str,
      help="Path to hf config.json of model to convert",
  )
  parser.add_argument("--pytorch_dump_folder_path", default=None, type=str, help="Path to the output PyTorch model.")
  parser.add_argument(
      "--push_to_hub",
      action="store_true",
      help="Whether to push the model and processor to the hub after converting",
  )
  parser.add_argument(
      "--model_hub_id",
      default="ybelkada/segment-anything",
      choices=choices,
      type=str,
      help="Path to hf config.json of model to convert",
  )

  lowercase_	 =						parser.parse_args()

  convert_sam_checkpoint(args.model_name, args.pytorch_dump_folder_path, args.push_to_hub, args.model_hub_id)











'''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 __A    (						A     ,     unittest.TestCase		):



 '''simple docstring'''







 __lowerCamelCase					: Dict    = 'hf-internal-testing/tiny-random-OnnxStableDiffusionPipeline'



 def       a__						(self				,     A=0 )					->  str:


      """simple docstring"""



      _a											=						floats_tensor((1, 3, 128, 128)				,     rng=random.Random(A ) )
      _a											=						np.random.RandomState(A )
      _a											=						{
          '''prompt''': '''A painting of a squirrel eating a burger''',
          '''image''': image,
          '''generator''': generator,
          '''num_inference_steps''': 3,
          '''strength''': 0.75,
          '''guidance_scale''': 7.5,
          '''output_type''': '''numpy''',
      }
      return inputs



 def       a__						(self )					->  str:


      """simple docstring"""



      _a											=						OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint				,     provider='''CPUExecutionProvider''' )
      pipe.set_progress_bar_config(disable=A )

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

      assert image.shape == (1, 128, 128, 3)
      _a											=						np.array([0.69643, 0.58484, 0.50314, 0.58760, 0.55368, 0.59643, 0.51529, 0.41217, 0.49087] )
      assert np.abs(image_slice - expected_slice ).max() < 1E-1



 def       a__						(self )					->  Tuple:


      """simple docstring"""



      _a											=						OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint				,     provider='''CPUExecutionProvider''' )
      _a											=						PNDMScheduler.from_config(pipe.scheduler.config				,     skip_prk_steps=A )
      pipe.set_progress_bar_config(disable=A )

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

      assert image.shape == (1, 128, 128, 3)
      _a											=						np.array([0.61737, 0.54642, 0.53183, 0.54465, 0.52742, 0.60525, 0.49969, 0.40655, 0.48154] )

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



 def       a__						(self )					->  int:


      """simple docstring"""



      _a											=						OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint				,     provider='''CPUExecutionProvider''' )
      _a											=						LMSDiscreteScheduler.from_config(pipe.scheduler.config )
      pipe.set_progress_bar_config(disable=A )

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

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

      assert image.shape == (1, 128, 128, 3)
      _a											=						np.array([0.52761, 0.59977, 0.49033, 0.49619, 0.54282, 0.50311, 0.47600, 0.40918, 0.45203] )

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



 def       a__						(self )					->  Dict:


      """simple docstring"""



      _a											=						OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint				,     provider='''CPUExecutionProvider''' )
      _a											=						EulerDiscreteScheduler.from_config(pipe.scheduler.config )
      pipe.set_progress_bar_config(disable=A )

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

      assert image.shape == (1, 128, 128, 3)
      _a											=						np.array([0.52911, 0.60004, 0.49229, 0.49805, 0.54502, 0.50680, 0.47777, 0.41028, 0.45304] )

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



 def       a__						(self )					->  Tuple:


      """simple docstring"""



      _a											=						OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint				,     provider='''CPUExecutionProvider''' )
      _a											=						EulerAncestralDiscreteScheduler.from_config(pipe.scheduler.config )
      pipe.set_progress_bar_config(disable=A )

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

      assert image.shape == (1, 128, 128, 3)
      _a											=						np.array([0.52911, 0.60004, 0.49229, 0.49805, 0.54502, 0.50680, 0.47777, 0.41028, 0.45304] )

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



 def       a__						(self )					->  List[str]:


      """simple docstring"""



      _a											=						OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint				,     provider='''CPUExecutionProvider''' )
      _a											=						DPMSolverMultistepScheduler.from_config(pipe.scheduler.config )
      pipe.set_progress_bar_config(disable=A )

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

      assert image.shape == (1, 128, 128, 3)
      _a											=						np.array([0.65331, 0.58277, 0.48204, 0.56059, 0.53665, 0.56235, 0.50969, 0.40009, 0.46552] )

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






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



 '''simple docstring'''







 @property
 def       a__						(self )					->  Tuple:


      """simple docstring"""



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



 @property
 def       a__						(self )					->  Any:


      """simple docstring"""



      _a											=						ort.SessionOptions()
      _a											=						False
      return options



 def       a__						(self )					->  str:


      """simple docstring"""



      _a											=						load_image(
          '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
          '''/img2img/sketch-mountains-input.jpg''' )
      _a											=						init_image.resize((768, 512) )
      # using the PNDM scheduler by default
      _a											=						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 )

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

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

      assert images.shape == (1, 512, 768, 3)
      _a											=						np.array([0.4909, 0.5059, 0.5372, 0.4623, 0.4876, 0.5049, 0.4820, 0.4956, 0.5019] )
      # TODO: lower the tolerance after finding the cause of onnxruntime reproducibility issues

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



 def       a__						(self )					->  Optional[int]:


      """simple docstring"""



      _a											=						load_image(
          '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
          '''/img2img/sketch-mountains-input.jpg''' )
      _a											=						init_image.resize((768, 512) )
      _a											=						LMSDiscreteScheduler.from_pretrained(
          '''runwayml/stable-diffusion-v1-5'''				,     subfolder='''scheduler'''				,     revision='''onnx''' )
      _a											=						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 )

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

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

      assert images.shape == (1, 512, 768, 3)
      _a											=						np.array([0.8043, 0.926, 0.9581, 0.8119, 0.8954, 0.913, 0.7209, 0.7463, 0.7431] )
      # TODO: lower the tolerance after finding the cause of onnxruntime reproducibility issues

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



