


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					(	__snake_case			,       unittest.TestCase	):
      UpperCamelCase										=							'''hf-internal-testing/tiny-random-OnnxStableDiffusionPipeline'''





      def 							_lowercase					(		self    :       Tuple       ,							__lowerCamelCase    :       Dict=0							)     -> List[Any]:






         """simple docstring"""







         UpperCAmelCase      					=  floats_tensor((1, 3, 1_2_8, 1_2_8)       ,							rng=random.Random(__lowerCamelCase							)							)
         UpperCAmelCase      					=  np.random.RandomState(__lowerCamelCase							)
         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 							_lowercase					(		self    :       Any							)     -> Dict:






         """simple docstring"""







         UpperCAmelCase      					=  OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint       ,							provider="""CPUExecutionProvider"""							)
         pipe.set_progress_bar_config(disable=__lowerCamelCase							)

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

         assert image.shape == (1, 1_2_8, 1_2_8, 3)
         UpperCAmelCase      					=  np.array([0.69_643, 0.58_484, 0.50_314, 0.58_760, 0.55_368, 0.59_643, 0.51_529, 0.41_217, 0.49_087]							)
         assert np.abs(image_slice - expected_slice							).max() < 1e-1





      def 							_lowercase					(		self    :       Tuple							)     -> str:






         """simple docstring"""







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

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

         assert image.shape == (1, 1_2_8, 1_2_8, 3)
         UpperCAmelCase      					=  np.array([0.61_737, 0.54_642, 0.53_183, 0.54_465, 0.52_742, 0.60_525, 0.49_969, 0.40_655, 0.48_154]							)

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





      def 							_lowercase					(		self    :       str							)     -> 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=__lowerCamelCase							)

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

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

         assert image.shape == (1, 1_2_8, 1_2_8, 3)
         UpperCAmelCase      					=  np.array([0.52_761, 0.59_977, 0.49_033, 0.49_619, 0.54_282, 0.50_311, 0.47_600, 0.40_918, 0.45_203]							)

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





      def 							_lowercase					(		self    :       List[Any]							)     -> Any:






         """simple docstring"""







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

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

         assert image.shape == (1, 1_2_8, 1_2_8, 3)
         UpperCAmelCase      					=  np.array([0.52_911, 0.60_004, 0.49_229, 0.49_805, 0.54_502, 0.50_680, 0.47_777, 0.41_028, 0.45_304]							)

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





      def 							_lowercase					(		self    :       int							)     -> str:






         """simple docstring"""







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

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

         assert image.shape == (1, 1_2_8, 1_2_8, 3)
         UpperCAmelCase      					=  np.array([0.52_911, 0.60_004, 0.49_229, 0.49_805, 0.54_502, 0.50_680, 0.47_777, 0.41_028, 0.45_304]							)

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





      def 							_lowercase					(		self    :       Any							)     -> Optional[Any]:






         """simple docstring"""







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

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

         assert image.shape == (1, 1_2_8, 1_2_8, 3)
         UpperCAmelCase      					=  np.array([0.65_331, 0.58_277, 0.48_204, 0.56_059, 0.53_665, 0.56_235, 0.50_969, 0.40_009, 0.46_552]							)

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






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





      @property
      def 							_lowercase					(		self    :       Dict							)     -> int:






         """simple docstring"""







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





      @property
      def 							_lowercase					(		self    :       Tuple							)     -> str:






         """simple docstring"""







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





      def 							_lowercase					(		self    :       Optional[int]							)     -> List[Any]:






         """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=__lowerCamelCase       ,							feature_extractor=__lowerCamelCase       ,							provider=self.gpu_provider       ,							sess_options=self.gpu_options       ,							)
         pipe.set_progress_bar_config(disable=__lowerCamelCase							)

         UpperCAmelCase      					=  """A fantasy landscape, trending on artstation"""

         UpperCAmelCase      					=  np.random.RandomState(0							)
         UpperCAmelCase      					=  pipe(
             prompt=__lowerCamelCase       ,							image=__lowerCamelCase       ,							strength=0.75       ,							guidance_scale=7.5       ,							num_inference_steps=1_0       ,							generator=__lowerCamelCase       ,							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.4_909, 0.5_059, 0.5_372, 0.4_623, 0.4_876, 0.5_049, 0.4_820, 0.4_956, 0.5_019]							)
         # TODO: lower the tolerance after finding the cause of onnxruntime reproducibility issues

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





      def 							_lowercase					(		self    :       Dict							)     -> Optional[Any]:






         """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=__lowerCamelCase       ,							safety_checker=__lowerCamelCase       ,							feature_extractor=__lowerCamelCase       ,							provider=self.gpu_provider       ,							sess_options=self.gpu_options       ,							)
         pipe.set_progress_bar_config(disable=__lowerCamelCase							)

         UpperCAmelCase      					=  """A fantasy landscape, trending on artstation"""

         UpperCAmelCase      					=  np.random.RandomState(0							)
         UpperCAmelCase      					=  pipe(
             prompt=__lowerCamelCase       ,							image=__lowerCamelCase       ,							strength=0.75       ,							guidance_scale=7.5       ,							num_inference_steps=2_0       ,							generator=__lowerCamelCase       ,							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.8_043, 0.926, 0.9_581, 0.8_119, 0.8_954, 0.913, 0.7_209, 0.7_463, 0.7_431]							)
         # TODO: lower the tolerance after finding the cause of onnxruntime reproducibility issues

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







import argparse
from pathlib import Path

import requests
import torch
from PIL import Image

from transformers import (
    RobertaTokenizer,
    TrOCRConfig,
    TrOCRForCausalLM,
    TrOCRProcessor,
    VisionEncoderDecoderModel,
    ViTConfig,
    ViTImageProcessor,
    ViTModel,
)
from transformers.utils import logging


logging.set_verbosity_info()
__a	 =      logging.get_logger(__name__)
def 	_UpperCamelCase					(    lowerCAmelCase_	,    lowerCAmelCase_ )    ->Union[str, Any]:
   UpperCAmelCase      					=  []
   for i in range(encoder_config.num_hidden_layers ):
      # encoder layers: output projection, 2 feedforward neural networks and 2 layernorms
      rename_keys.append(
          (F"""encoder.deit.blocks.{i}.norm1.weight""", F"""encoder.encoder.layer.{i}.layernorm_before.weight""") )
      rename_keys.append((F"""encoder.deit.blocks.{i}.norm1.bias""", F"""encoder.encoder.layer.{i}.layernorm_before.bias""") )
      rename_keys.append(
          (F"""encoder.deit.blocks.{i}.attn.proj.weight""", F"""encoder.encoder.layer.{i}.attention.output.dense.weight""") )
      rename_keys.append(
          (F"""encoder.deit.blocks.{i}.attn.proj.bias""", F"""encoder.encoder.layer.{i}.attention.output.dense.bias""") )
      rename_keys.append(
          (F"""encoder.deit.blocks.{i}.norm2.weight""", F"""encoder.encoder.layer.{i}.layernorm_after.weight""") )
      rename_keys.append((F"""encoder.deit.blocks.{i}.norm2.bias""", F"""encoder.encoder.layer.{i}.layernorm_after.bias""") )
      rename_keys.append(
          (F"""encoder.deit.blocks.{i}.mlp.fc1.weight""", F"""encoder.encoder.layer.{i}.intermediate.dense.weight""") )
      rename_keys.append(
          (F"""encoder.deit.blocks.{i}.mlp.fc1.bias""", F"""encoder.encoder.layer.{i}.intermediate.dense.bias""") )
      rename_keys.append(
          (F"""encoder.deit.blocks.{i}.mlp.fc2.weight""", F"""encoder.encoder.layer.{i}.output.dense.weight""") )
      rename_keys.append((F"""encoder.deit.blocks.{i}.mlp.fc2.bias""", F"""encoder.encoder.layer.{i}.output.dense.bias""") )

   # cls token, position embeddings and patch embeddings of encoder
   rename_keys.extend(
       [
           ("""encoder.deit.cls_token""", """encoder.embeddings.cls_token"""),
           ("""encoder.deit.pos_embed""", """encoder.embeddings.position_embeddings"""),
           ("""encoder.deit.patch_embed.proj.weight""", """encoder.embeddings.patch_embeddings.projection.weight"""),
           ("""encoder.deit.patch_embed.proj.bias""", """encoder.embeddings.patch_embeddings.projection.bias"""),
           ("""encoder.deit.norm.weight""", """encoder.layernorm.weight"""),
           ("""encoder.deit.norm.bias""", """encoder.layernorm.bias"""),
       ] )

   return rename_keys
def 	_UpperCamelCase					(    lowerCAmelCase_	,    lowerCAmelCase_ )    ->Any:
   for i in range(encoder_config.num_hidden_layers ):
      # queries, keys and values (only weights, no biases)
      UpperCAmelCase      					=  state_dict.pop(F"""encoder.deit.blocks.{i}.attn.qkv.weight""" )

      UpperCAmelCase      					=  in_proj_weight[
          : encoder_config.hidden_size, :
      ]
      UpperCAmelCase      					=  in_proj_weight[
          encoder_config.hidden_size : encoder_config.hidden_size * 2, :
      ]
      UpperCAmelCase      					=  in_proj_weight[
          -encoder_config.hidden_size :, :
      ]
def 	_UpperCamelCase					(    lowerCAmelCase_	,    lowerCAmelCase_	,    lowerCAmelCase_ )    ->Optional[Any]:
   UpperCAmelCase      					=  dct.pop(lowerCAmelCase_ )
   UpperCAmelCase      					=  val
def 	_UpperCamelCase					(    lowerCAmelCase_ )    ->Any:
   if "handwritten" in checkpoint_url:
      UpperCAmelCase      					=  """https://fki.tic.heia-fr.ch/static/img/a01-122-02-00.jpg"""  # industry
      # url = "https://fki.tic.heia-fr.ch/static/img/a01-122-02-12.jpg" # have
      # url = "https://fki.tic.heia-fr.ch/static/img/a01-122-02-10.jpg" # let
      # url = "https://fki.tic.heia-fr.ch/static/img/a01-122-02.jpg"  #
      # url = "https://fki.tic.heia-fr.ch/static/img/a01-122.jpg"
   elif "printed" in checkpoint_url or "stage1" in checkpoint_url:
      UpperCAmelCase      					=  """https://www.researchgate.net/profile/Dinh-Sang/publication/338099565/figure/fig8/AS:840413229350922@1577381536857/An-receipt-example-in-the-SROIE-2019-dataset_Q640.jpg"""
   UpperCAmelCase      					=  Image.open(requests.get(lowerCAmelCase_	,    stream=lowerCAmelCase_ ).raw ).convert("""RGB""" )
   return im
@torch.no_grad()
def 	_UpperCamelCase					(    lowerCAmelCase_	,    lowerCAmelCase_ )    ->Optional[int]:
   UpperCAmelCase      					=  ViTConfig(image_size=3_8_4	,    qkv_bias=lowerCAmelCase_ )
   UpperCAmelCase      					=  TrOCRConfig()

   # size of the architecture
   if "base" in checkpoint_url:
      UpperCAmelCase      					=  7_6_8
   elif "large" in checkpoint_url:
      # use ViT-large encoder
      UpperCAmelCase      					=  1_0_2_4
      UpperCAmelCase      					=  4_0_9_6
      UpperCAmelCase      					=  2_4
      UpperCAmelCase      					=  1_6
      UpperCAmelCase      					=  1_0_2_4
   else:
      raise ValueError("""Should either find 'base' or 'large' in checkpoint URL""" )

   # the large-printed + stage1 checkpoints uses sinusoidal position embeddings, no layernorm afterwards
   if "large-printed" in checkpoint_url or "stage1" in checkpoint_url:
      UpperCAmelCase      					=  False
      UpperCAmelCase      					=  """relu"""
      UpperCAmelCase      					=  1_0_2_4
      UpperCAmelCase      					=  True
      UpperCAmelCase      					=  False
      UpperCAmelCase      					=  False

   # load HuggingFace model
   UpperCAmelCase      					=  ViTModel(lowerCAmelCase_	,    add_pooling_layer=lowerCAmelCase_ )
   UpperCAmelCase      					=  TrOCRForCausalLM(lowerCAmelCase_ )
   UpperCAmelCase      					=  VisionEncoderDecoderModel(encoder=lowerCAmelCase_	,    decoder=lowerCAmelCase_ )
   model.eval()

   # load state_dict of original model, rename some keys
   UpperCAmelCase      					=  torch.hub.load_state_dict_from_url(lowerCAmelCase_	,    map_location="""cpu"""	,    check_hash=lowerCAmelCase_ )["""model"""]

   UpperCAmelCase      					=  create_rename_keys(lowerCAmelCase_	,    lowerCAmelCase_ )
   for src, dest in rename_keys:
      rename_key(lowerCAmelCase_	,    lowerCAmelCase_	,    lowerCAmelCase_ )
   read_in_q_k_v(lowerCAmelCase_	,    lowerCAmelCase_ )

   # remove parameters we don't need
   del state_dict["encoder.deit.head.weight"]
   del state_dict["encoder.deit.head.bias"]
   del state_dict["decoder.version"]

   # add prefix to decoder keys
   for key, val in state_dict.copy().items():
      UpperCAmelCase      					=  state_dict.pop(lowerCAmelCase_ )
      if key.startswith("""decoder""" ) and "output_projection" not in key:
         UpperCAmelCase      					=  val
      else:
         UpperCAmelCase      					=  val

    # load state dict
   model.load_state_dict(lowerCAmelCase_ )

   # Check outputs on an image
   UpperCAmelCase      					=  ViTImageProcessor(size=encoder_config.image_size )
   UpperCAmelCase      					=  RobertaTokenizer.from_pretrained("""roberta-large""" )
   UpperCAmelCase      					=  TrOCRProcessor(lowerCAmelCase_	,    lowerCAmelCase_ )

   UpperCAmelCase      					=  processor(images=prepare_img(lowerCAmelCase_ )	,    return_tensors="""pt""" ).pixel_values

   # verify logits
   UpperCAmelCase      					=  torch.tensor([[model.config.decoder.decoder_start_token_id]] )
   UpperCAmelCase      					=  model(pixel_values=lowerCAmelCase_	,    decoder_input_ids=lowerCAmelCase_ )
   UpperCAmelCase      					=  outputs.logits

   UpperCAmelCase      					=  torch.Size([1, 1, 5_0_2_6_5] )
   if "trocr-base-handwritten" in checkpoint_url:
      UpperCAmelCase      					=  torch.tensor(
          [-1.4502, -4.6683, -0.5347, -2.9291, 9.1435, -3.0571, 8.9764, 1.7560, 8.7358, -1.5311] )
   elif "trocr-large-handwritten" in checkpoint_url:
      UpperCAmelCase      					=  torch.tensor(
          [-2.6437, -1.3129, -2.2596, -5.3455, 6.3539, 1.7604, 5.4991, 1.4702, 5.6113, 2.0170] )
   elif "trocr-base-printed" in checkpoint_url:
      UpperCAmelCase      					=  torch.tensor(
          [-5.6816, -5.8388, 1.1398, -6.9034, 6.8505, -2.4393, 1.2284, -1.0232, -1.9661, -3.9210] )
   elif "trocr-large-printed" in checkpoint_url:
      UpperCAmelCase      					=  torch.tensor(
          [-6.0162, -7.0959, 4.4155, -5.1063, 7.0468, -3.1631, 2.6466, -0.3081, -0.8106, -1.7535] )

   if "stage1" not in checkpoint_url:
      assert logits.shape == expected_shape, "Shape of logits not as expected"
      assert torch.allclose(logits[0, 0, :1_0]	,    lowerCAmelCase_	,    atol=1e-3 ), "First elements of logits not as expected"

   Path(lowerCAmelCase_ ).mkdir(exist_ok=lowerCAmelCase_ )
   print(F"""Saving model to {pytorch_dump_folder_path}""" )
   model.save_pretrained(lowerCAmelCase_ )
   print(F"""Saving processor to {pytorch_dump_folder_path}""" )
   processor.save_pretrained(lowerCAmelCase_ )


if __name__ == "__main__":
 __a	 =      argparse.ArgumentParser()

 parser.add_argument(
     """--checkpoint_url""",
     default="""https://layoutlm.blob.core.windows.net/trocr/model_zoo/fairseq/trocr-base-handwritten.pt""",
     type=str,
     help="""URL to the original PyTorch checkpoint (.pth file).""",
 )
 parser.add_argument(
     """--pytorch_dump_folder_path""", default=None, type=str, help="""Path to the folder to output PyTorch model."""
 )
 __a	 =      parser.parse_args()
 convert_tr_ocr_checkpoint(args.checkpoint_url, args.pytorch_dump_folder_path)



