

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

  '''simple docstring'''





  __lowerCAmelCase       :			List[Any]     	= "hf-internal-testing/tiny-random-OnnxStableDiffusionPipeline"
  def     snake_case_      (  self ,			a_=0  )				->					str:


    """simple docstring"""
    UpperCAmelCase    				=						floats_tensor((1, 3, 1_2_8, 1_2_8) ,			rng=random.Random(a_  )  )
    UpperCAmelCase    				=						np.random.RandomState(a_  )
    UpperCAmelCase    				=						{
        '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     snake_case_      (  self  )				->					str:


    """simple docstring"""
    UpperCAmelCase    				=						OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint ,			provider='CPUExecutionProvider'  )
    pipe.set_progress_bar_config(disable=a_  )

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

    assert image.shape == (1, 1_2_8, 1_2_8, 3)
    UpperCAmelCase    				=						np.array([0.6_9643, 0.5_8484, 0.5_0314, 0.5_8760, 0.5_5368, 0.5_9643, 0.5_1529, 0.4_1217, 0.4_9087]  )
    assert np.abs(image_slice - expected_slice  ).max() < 1E-1
  def     snake_case_      (  self  )				->					str:


    """simple docstring"""
    UpperCAmelCase    				=						OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint ,			provider='CPUExecutionProvider'  )
    UpperCAmelCase    				=						PNDMScheduler.from_config(pipe.scheduler.config ,			skip_prk_steps=a_  )
    pipe.set_progress_bar_config(disable=a_  )

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

    assert image.shape == (1, 1_2_8, 1_2_8, 3)
    UpperCAmelCase    				=						np.array([0.6_1737, 0.5_4642, 0.5_3183, 0.5_4465, 0.5_2742, 0.6_0525, 0.4_9969, 0.4_0655, 0.4_8154]  )

    assert np.abs(image_slice.flatten() - expected_slice  ).max() < 1E-1
  def     snake_case_      (  self  )				->					int:


    """simple docstring"""
    UpperCAmelCase    				=						OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint ,			provider='CPUExecutionProvider'  )
    UpperCAmelCase    				=						LMSDiscreteScheduler.from_config(pipe.scheduler.config  )
    pipe.set_progress_bar_config(disable=a_  )

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

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

    assert image.shape == (1, 1_2_8, 1_2_8, 3)
    UpperCAmelCase    				=						np.array([0.5_2761, 0.5_9977, 0.4_9033, 0.4_9619, 0.5_4282, 0.5_0311, 0.4_7600, 0.4_0918, 0.4_5203]  )

    assert np.abs(image_slice.flatten() - expected_slice  ).max() < 1E-1
  def     snake_case_      (  self  )				->					Dict:


    """simple docstring"""
    UpperCAmelCase    				=						OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint ,			provider='CPUExecutionProvider'  )
    UpperCAmelCase    				=						EulerDiscreteScheduler.from_config(pipe.scheduler.config  )
    pipe.set_progress_bar_config(disable=a_  )

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

    assert image.shape == (1, 1_2_8, 1_2_8, 3)
    UpperCAmelCase    				=						np.array([0.5_2911, 0.6_0004, 0.4_9229, 0.4_9805, 0.5_4502, 0.5_0680, 0.4_7777, 0.4_1028, 0.4_5304]  )

    assert np.abs(image_slice.flatten() - expected_slice  ).max() < 1E-1
  def     snake_case_      (  self  )				->					Any:


    """simple docstring"""
    UpperCAmelCase    				=						OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint ,			provider='CPUExecutionProvider'  )
    UpperCAmelCase    				=						EulerAncestralDiscreteScheduler.from_config(pipe.scheduler.config  )
    pipe.set_progress_bar_config(disable=a_  )

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

    assert image.shape == (1, 1_2_8, 1_2_8, 3)
    UpperCAmelCase    				=						np.array([0.5_2911, 0.6_0004, 0.4_9229, 0.4_9805, 0.5_4502, 0.5_0680, 0.4_7777, 0.4_1028, 0.4_5304]  )

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





  def     snake_case_      (  self  )				->					int:


    """simple docstring"""
    UpperCAmelCase    				=						OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint ,			provider='CPUExecutionProvider'  )
    UpperCAmelCase    				=						DPMSolverMultistepScheduler.from_config(pipe.scheduler.config  )
    pipe.set_progress_bar_config(disable=a_  )

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

    assert image.shape == (1, 1_2_8, 1_2_8, 3)
    UpperCAmelCase    				=						np.array([0.6_5331, 0.5_8277, 0.4_8204, 0.5_6059, 0.5_3665, 0.5_6235, 0.5_0969, 0.4_0009, 0.4_6552]  )

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




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

  '''simple docstring'''





  @property
  def     snake_case_      (  self  )				->					Tuple:


    """simple docstring"""
    return (
        "CUDAExecutionProvider",
        {
            "gpu_mem_limit": "15000000000",  # 15GB
            "arena_extend_strategy": "kSameAsRequested",
        },
    )
  @property
  def     snake_case_      (  self  )				->					Dict:


    """simple docstring"""
    UpperCAmelCase    				=						ort.SessionOptions()
    UpperCAmelCase    				=						False
    return options
  def     snake_case_      (  self  )				->					Optional[int]:


    """simple docstring"""
    UpperCAmelCase    				=						load_image(
        'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
        '/img2img/sketch-mountains-input.jpg'  )
    UpperCAmelCase    				=						init_image.resize((7_6_8, 5_1_2)  )
    # using the PNDM scheduler by default
    UpperCAmelCase    				=						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_  )

    UpperCAmelCase    				=						'A fantasy landscape, trending on artstation'

    UpperCAmelCase    				=						np.random.RandomState(0  )
    UpperCAmelCase    				=						pipe(
        prompt=a_ ,			image=a_ ,			strength=0.75 ,			guidance_scale=7.5 ,			num_inference_steps=1_0 ,			generator=a_ ,			output_type='np' ,			)
    UpperCAmelCase    				=						output.images
    UpperCAmelCase    				=						images[0, 2_5_5:2_5_8, 3_8_3:3_8_6, -1]

    assert images.shape == (1, 5_1_2, 7_6_8, 3)
    UpperCAmelCase    				=						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     snake_case_      (  self  )				->					int:


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

    UpperCAmelCase    				=						'A fantasy landscape, trending on artstation'

    UpperCAmelCase    				=						np.random.RandomState(0  )
    UpperCAmelCase    				=						pipe(
        prompt=a_ ,			image=a_ ,			strength=0.75 ,			guidance_scale=7.5 ,			num_inference_steps=2_0 ,			generator=a_ ,			output_type='np' ,			)
    UpperCAmelCase    				=						output.images
    UpperCAmelCase    				=						images[0, 2_5_5:2_5_8, 3_8_3:3_8_6, -1]

    assert images.shape == (1, 5_1_2, 7_6_8, 3)
    UpperCAmelCase    				=						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





'''simple docstring'''




import argparse
import json
from dataclasses import dataclass, field
from functools import partial
from pathlib import Path
from typing import List

import timm
import torch
import torch.nn as nn
from huggingface_hub import hf_hub_download
from torch import Tensor

from transformers import AutoImageProcessor, ResNetConfig, ResNetForImageClassification
from transformers.utils import logging


logging.set_verbosity_info()
_a					:			List[Any]							=    logging.get_logger()



@dataclass
class 							lowercase_						:

  '''simple docstring'''





  __lowerCAmelCase       :			nn.Module
  __lowerCAmelCase       :			List[nn.Module]     	= field(default_factory=a			)
  __lowerCAmelCase       :			list     	= field(default_factory=a			)
  def     snake_case_      (  self ,			a_ ,			a_ ,			a_  )				->					Tuple:


    """simple docstring"""
    UpperCAmelCase    				=						len(list(m.modules()  )  ) == 1 or isinstance(a_ ,			nn.Convad  ) or isinstance(a_ ,			nn.BatchNormad  )
    if has_not_submodules:
      self.traced.append(a_  )
  def __call__(  self ,			a_  )				->					Union[str, Any]:


    """simple docstring"""
    for m in self.module.modules():
      self.handles.append(m.register_forward_hook(self._forward_hook  )  )
    self.module(a_  )
    [x.remove() for x in self.handles]
    return self





  @property
  def     snake_case_      (  self  )				->					int:


    """simple docstring"""
    # check the len of the state_dict keys to see if we have learnable params
    return list(filter(lambda a_  : len(list(x.state_dict().keys()  )  ) > 0 ,			self.traced  )  )




@dataclass
class 							lowercase_						:

  '''simple docstring'''





  __lowerCAmelCase       :			nn.Module
  __lowerCAmelCase       :			nn.Module
  __lowerCAmelCase       :			int     	= 0
  __lowerCAmelCase       :			List     	= field(default_factory=a			)
  __lowerCAmelCase       :			List     	= field(default_factory=a			)
  def __call__(  self ,			a_  )				->					Dict:


    """simple docstring"""
    UpperCAmelCase    				=						Tracker(self.dest  )(a_  ).parametrized
    UpperCAmelCase    				=						Tracker(self.src  )(a_  ).parametrized

    UpperCAmelCase    				=						list(filter(lambda a_  : type(a_  ) not in self.src_skip ,			a_  )  )
    UpperCAmelCase    				=						list(filter(lambda a_  : type(a_  ) not in self.dest_skip ,			a_  )  )

    if len(a_  ) != len(a_  ):
      raise Exception(
          F'''Numbers of operations are different. Source module has {len(a_  )} operations while'''
          F''' destination module has {len(a_  )}.'''  )

    for dest_m, src_m in zip(a_ ,			a_  ):
      dest_m.load_state_dict(src_m.state_dict()  )
      if self.verbose == 1:
        print(F'''Transfered from={src_m} to={dest_m}'''  )




def    lowerCamelCase__		(    SCREAMING_SNAKE_CASE   : str							, SCREAMING_SNAKE_CASE   : ResNetConfig							, SCREAMING_SNAKE_CASE   : Path							, SCREAMING_SNAKE_CASE   : bool = True    ):
  print(f'''Converting {name}...'''    )
  with torch.no_grad():
    UpperCAmelCase    				=						timm.create_model(SCREAMING_SNAKE_CASE							, pretrained=SCREAMING_SNAKE_CASE    ).eval()
    UpperCAmelCase    				=						ResNetForImageClassification(SCREAMING_SNAKE_CASE    ).eval()
    UpperCAmelCase    				=						ModuleTransfer(src=SCREAMING_SNAKE_CASE							, dest=SCREAMING_SNAKE_CASE    )
    UpperCAmelCase    				=						torch.randn((1, 3, 224, 224)    )
    module_transfer(SCREAMING_SNAKE_CASE    )

  assert torch.allclose(from_model(SCREAMING_SNAKE_CASE    )							, our_model(SCREAMING_SNAKE_CASE    ).logits    ), "The model logits don't match the original one."

  UpperCAmelCase    				=						f'''resnet{'-'.join(name.split('resnet'    )    )}'''
  print(SCREAMING_SNAKE_CASE    )

  if push_to_hub:
    our_model.push_to_hub(
        repo_path_or_name=save_directory / checkpoint_name							, commit_message='Add model'							, use_temp_dir=SCREAMING_SNAKE_CASE							, )

    # we can use the convnext one
    UpperCAmelCase    				=						AutoImageProcessor.from_pretrained('facebook/convnext-base-224-22k-1k'    )
    image_processor.push_to_hub(
        repo_path_or_name=save_directory / checkpoint_name							, commit_message='Add image processor'							, use_temp_dir=SCREAMING_SNAKE_CASE							, )

    print(f'''Pushed {checkpoint_name}'''    )


def    lowerCamelCase__		(    SCREAMING_SNAKE_CASE   : Path							, SCREAMING_SNAKE_CASE   : str = None							, SCREAMING_SNAKE_CASE   : bool = True    ):
  UpperCAmelCase    				=						'imagenet-1k-id2label.json'
  UpperCAmelCase    				=						1000
  UpperCAmelCase    				=						(1, num_labels)

  UpperCAmelCase    				=						'huggingface/label-files'
  UpperCAmelCase    				=						num_labels
  UpperCAmelCase    				=						json.load(open(hf_hub_download(SCREAMING_SNAKE_CASE							, SCREAMING_SNAKE_CASE							, repo_type='dataset'    )							, 'r'    )    )
  UpperCAmelCase    				=						{int(SCREAMING_SNAKE_CASE    ): v for k, v in idalabel.items()}

  UpperCAmelCase    				=						idalabel
  UpperCAmelCase    				=						{v: k for k, v in idalabel.items()}

  UpperCAmelCase    				=						partial(SCREAMING_SNAKE_CASE							, num_labels=SCREAMING_SNAKE_CASE							, idalabel=SCREAMING_SNAKE_CASE							, labelaid=SCREAMING_SNAKE_CASE    )

  UpperCAmelCase    				=						{
      'resnet18': ImageNetPreTrainedConfig(
          depths=[2, 2, 2, 2]							, hidden_sizes=[64, 128, 256, 512]							, layer_type='basic'    ),
      'resnet26': ImageNetPreTrainedConfig(
          depths=[2, 2, 2, 2]							, hidden_sizes=[256, 512, 1024, 2048]							, layer_type='bottleneck'    ),
      'resnet34': ImageNetPreTrainedConfig(
          depths=[3, 4, 6, 3]							, hidden_sizes=[64, 128, 256, 512]							, layer_type='basic'    ),
      'resnet50': ImageNetPreTrainedConfig(
          depths=[3, 4, 6, 3]							, hidden_sizes=[256, 512, 1024, 2048]							, layer_type='bottleneck'    ),
      'resnet101': ImageNetPreTrainedConfig(
          depths=[3, 4, 23, 3]							, hidden_sizes=[256, 512, 1024, 2048]							, layer_type='bottleneck'    ),
      'resnet152': ImageNetPreTrainedConfig(
          depths=[3, 8, 36, 3]							, hidden_sizes=[256, 512, 1024, 2048]							, layer_type='bottleneck'    ),
  }

  if model_name:
    convert_weight_and_push(SCREAMING_SNAKE_CASE							, names_to_config[model_name]							, SCREAMING_SNAKE_CASE							, SCREAMING_SNAKE_CASE    )
  else:
    for model_name, config in names_to_config.items():
      convert_weight_and_push(SCREAMING_SNAKE_CASE							, SCREAMING_SNAKE_CASE							, SCREAMING_SNAKE_CASE							, SCREAMING_SNAKE_CASE    )
  return config, expected_shape


if __name__ == "__main__":
  _a					:			List[str]							=    argparse.ArgumentParser()
  # Required parameters
  parser.add_argument(
      '--model_name',
      default=None,
      type=str,
      help=(
          'The name of the model you wish to convert, it must be one of the supported resnet* architecture,'
          ' currently: resnet18,26,34,50,101,152. If `None`, all of them will the converted.'
      ),
  )
  parser.add_argument(
      '--pytorch_dump_folder_path',
      default=None,
      type=Path,
      required=True,
      help='Path to the output PyTorch model directory.',
  )
  parser.add_argument(
      '--push_to_hub',
      default=True,
      type=bool,
      required=False,
      help='If True, push model and image processor to the hub.',
  )

  _a					:			Tuple							=    parser.parse_args()
  _a					:			Path							=    args.pytorch_dump_folder_path
  pytorch_dump_folder_path.mkdir(exist_ok=True, parents=True)
  convert_weights_and_push(pytorch_dump_folder_path, args.model_name, args.push_to_hub)


