






"""simple docstring"""



import argparse
import json
from pathlib import Path

import requests
import torch
from huggingface_hub import hf_hub_download
from PIL import Image

from transformers import (
    BertTokenizer,
    ViltConfig,
    ViltForImageAndTextRetrieval,
    ViltForImagesAndTextClassification,
    ViltForMaskedLM,
    ViltForQuestionAnswering,
    ViltImageProcessor,
    ViltProcessor,
)
from transformers.utils import logging


logging.set_verbosity_info()
A__       :				Optional[int] 					=					logging.get_logger(__name__)

def  _snake_case				(     lowerCamelCase__	:    Optional[int]				,						lowerCamelCase__	:    List[str]=False				,						lowerCamelCase__	:    Any=False				,						lowerCamelCase__	:    Tuple=False  )				->	str:
							lowerCamelCase_       :							Optional[int]									=[]
							for i in range(config.num_hidden_layers  ):
														# encoder layers: output projection, 2 feedforward neural networks and 2 layernorms
														rename_keys.append((F"""transformer.blocks.{i}.norm1.weight""", F"""vilt.encoder.layer.{i}.layernorm_before.weight""")  )
														rename_keys.append((F"""transformer.blocks.{i}.norm1.bias""", F"""vilt.encoder.layer.{i}.layernorm_before.bias""")  )
														rename_keys.append(
														    (F"""transformer.blocks.{i}.attn.proj.weight""", F"""vilt.encoder.layer.{i}.attention.output.dense.weight""")  )
														rename_keys.append(
														    (F"""transformer.blocks.{i}.attn.proj.bias""", F"""vilt.encoder.layer.{i}.attention.output.dense.bias""")  )
														rename_keys.append((F"""transformer.blocks.{i}.norm2.weight""", F"""vilt.encoder.layer.{i}.layernorm_after.weight""")  )
														rename_keys.append((F"""transformer.blocks.{i}.norm2.bias""", F"""vilt.encoder.layer.{i}.layernorm_after.bias""")  )
														rename_keys.append(
														    (F"""transformer.blocks.{i}.mlp.fc1.weight""", F"""vilt.encoder.layer.{i}.intermediate.dense.weight""")  )
														rename_keys.append((F"""transformer.blocks.{i}.mlp.fc1.bias""", F"""vilt.encoder.layer.{i}.intermediate.dense.bias""")  )
														rename_keys.append((F"""transformer.blocks.{i}.mlp.fc2.weight""", F"""vilt.encoder.layer.{i}.output.dense.weight""")  )
														rename_keys.append((F"""transformer.blocks.{i}.mlp.fc2.bias""", F"""vilt.encoder.layer.{i}.output.dense.bias""")  )

							# embeddings
							rename_keys.extend(
							    [
							        # text embeddings
							        ("text_embeddings.word_embeddings.weight", "vilt.embeddings.text_embeddings.word_embeddings.weight"),
							        (
							            "text_embeddings.position_embeddings.weight",
							            "vilt.embeddings.text_embeddings.position_embeddings.weight",
							        ),
							        ("text_embeddings.position_ids", "vilt.embeddings.text_embeddings.position_ids"),
							        (
							            "text_embeddings.token_type_embeddings.weight",
							            "vilt.embeddings.text_embeddings.token_type_embeddings.weight",
							        ),
							        ("text_embeddings.LayerNorm.weight", "vilt.embeddings.text_embeddings.LayerNorm.weight"),
							        ("text_embeddings.LayerNorm.bias", "vilt.embeddings.text_embeddings.LayerNorm.bias"),
							        # patch embeddings
							        ("transformer.cls_token", "vilt.embeddings.cls_token"),
							        ("transformer.patch_embed.proj.weight", "vilt.embeddings.patch_embeddings.projection.weight"),
							        ("transformer.patch_embed.proj.bias", "vilt.embeddings.patch_embeddings.projection.bias"),
							        ("transformer.pos_embed", "vilt.embeddings.position_embeddings"),
							        # token type embeddings
							        ("token_type_embeddings.weight", "vilt.embeddings.token_type_embeddings.weight"),
							    ]  )

							# final layernorm + pooler
							rename_keys.extend(
							    [
							        ("transformer.norm.weight", "vilt.layernorm.weight"),
							        ("transformer.norm.bias", "vilt.layernorm.bias"),
							        ("pooler.dense.weight", "vilt.pooler.dense.weight"),
							        ("pooler.dense.bias", "vilt.pooler.dense.bias"),
							    ]  )

							# classifier head(s)
							if vqa_model:
														# classification head
														rename_keys.extend(
														    [
														        ("vqa_classifier.0.weight", "classifier.0.weight"),
														        ("vqa_classifier.0.bias", "classifier.0.bias"),
														        ("vqa_classifier.1.weight", "classifier.1.weight"),
														        ("vqa_classifier.1.bias", "classifier.1.bias"),
														        ("vqa_classifier.3.weight", "classifier.3.weight"),
														        ("vqa_classifier.3.bias", "classifier.3.bias"),
														    ]  )
							elif nlvr_model:
														# classification head
														rename_keys.extend(
														    [
														        ("nlvr2_classifier.0.weight", "classifier.0.weight"),
														        ("nlvr2_classifier.0.bias", "classifier.0.bias"),
														        ("nlvr2_classifier.1.weight", "classifier.1.weight"),
														        ("nlvr2_classifier.1.bias", "classifier.1.bias"),
														        ("nlvr2_classifier.3.weight", "classifier.3.weight"),
														        ("nlvr2_classifier.3.bias", "classifier.3.bias"),
														    ]  )
							else:
														pass

							return rename_keys

def  _snake_case				(     lowerCamelCase__	:    Tuple				,						lowerCamelCase__	:    int  )				->	int:
							for i in range(config.num_hidden_layers  ):
														lowerCamelCase_       :							Optional[Any]									='''vilt.'''
														# read in weights + bias of input projection layer (in timm, this is a single matrix + bias)
														lowerCamelCase_       :							Optional[Any]									=state_dict.pop(F"""transformer.blocks.{i}.attn.qkv.weight"""  )
														lowerCamelCase_       :							Any									=state_dict.pop(F"""transformer.blocks.{i}.attn.qkv.bias"""  )
														# next, add query, keys and values (in that order) to the state dict
														lowerCamelCase_       :							Optional[Any]									=in_proj_weight[
														    : config.hidden_size, :
														]
														lowerCamelCase_       :							List[str]									=in_proj_bias[: config.hidden_size]
														lowerCamelCase_       :							int									=in_proj_weight[
														    config.hidden_size : config.hidden_size * 2, :
														]
														lowerCamelCase_       :							int									=in_proj_bias[
														    config.hidden_size : config.hidden_size * 2
														]
														lowerCamelCase_       :							Optional[int]									=in_proj_weight[
														    -config.hidden_size :, :
														]
														lowerCamelCase_       :							Optional[Any]									=in_proj_bias[-config.hidden_size :]

def  _snake_case				(     lowerCamelCase__	:    List[str]  )				->	int:
							lowerCamelCase_       :							int									=['''head.weight''', '''head.bias''']
							for k in ignore_keys:
														state_dict.pop(a_				,						a_  )

def  _snake_case				(     lowerCamelCase__	:    Union[str, Any]				,						lowerCamelCase__	:    Tuple				,						lowerCamelCase__	:    Union[str, Any]  )				->	Tuple:
							lowerCamelCase_       :							str									=dct.pop(a_  )
							lowerCamelCase_       :							List[str]									=val

@torch.no_grad()
def  _snake_case				(     lowerCamelCase__	:    List[Any]				,						lowerCamelCase__	:    List[Any]  )				->	int:
							lowerCamelCase_       :							Tuple									=ViltConfig(image_size=384				,						patch_size=32				,						tie_word_embeddings=a_  )
							lowerCamelCase_       :							Tuple									=False
							lowerCamelCase_       :							List[str]									=False
							lowerCamelCase_       :							int									=False
							lowerCamelCase_       :							Optional[Any]									=False
							if "vqa" in checkpoint_url:
														lowerCamelCase_       :							Optional[int]									=True
														lowerCamelCase_       :							Tuple									=3_129
														lowerCamelCase_       :							List[str]									='''huggingface/label-files'''
														lowerCamelCase_       :							Optional[Any]									='''vqa2-id2label.json'''
														lowerCamelCase_       :							Dict									=json.load(open(hf_hub_download(a_				,						a_				,						repo_type="dataset"  )				,						"r"  )  )
														lowerCamelCase_       :							Dict									={int(a_  ): v for k, v in idalabel.items()}
														lowerCamelCase_       :							Optional[Any]									=idalabel
														lowerCamelCase_       :							List[str]									={v: k for k, v in idalabel.items()}
														lowerCamelCase_       :							Optional[Any]									=ViltForQuestionAnswering(a_  )
							elif "nlvr" in checkpoint_url:
														lowerCamelCase_       :							Any									=True
														lowerCamelCase_       :							Union[str, Any]									=2
														lowerCamelCase_       :							Union[str, Any]									={0: '''False''', 1: '''True'''}
														lowerCamelCase_       :							Tuple									={v: k for k, v in config.idalabel.items()}
														lowerCamelCase_       :							Optional[Any]									=3
														lowerCamelCase_       :							Union[str, Any]									=ViltForImagesAndTextClassification(a_  )
							elif "irtr" in checkpoint_url:
														lowerCamelCase_       :							Union[str, Any]									=True
														lowerCamelCase_       :							Optional[Any]									=ViltForImageAndTextRetrieval(a_  )
							elif "mlm_itm" in checkpoint_url:
														lowerCamelCase_       :							Dict									=True
														lowerCamelCase_       :							str									=ViltForMaskedLM(a_  )
							else:
														raise ValueError("Unknown model type"  )

							# load state_dict of original model, remove and rename some keys
							lowerCamelCase_       :							Union[str, Any]									=torch.hub.load_state_dict_from_url(a_				,						map_location="cpu"  )['''state_dict''']
							lowerCamelCase_       :							Optional[Any]									=create_rename_keys(a_				,						a_				,						a_				,						a_  )
							for src, dest in rename_keys:
														rename_key(a_				,						a_				,						a_  )
							read_in_q_k_v(a_				,						a_  )
							if mlm_model or irtr_model:
														lowerCamelCase_       :							Dict									=['''itm_score.fc.weight''', '''itm_score.fc.bias''']
														for k in ignore_keys:
																					state_dict.pop(a_				,						a_  )

    # load state dict into HuggingFace model
							model.eval()
							if mlm_model:
														lowerCamelCase_       :							Tuple									=model.load_state_dict(a_				,						strict=a_  )
														assert missing_keys == ["mlm_score.decoder.bias"]
							else:
														model.load_state_dict(a_  )

							# Define processor
							lowerCamelCase_       :							int									=ViltImageProcessor(size=384  )
							lowerCamelCase_       :							List[Any]									=BertTokenizer.from_pretrained("bert-base-uncased"  )
							lowerCamelCase_       :							List[Any]									=ViltProcessor(a_				,						a_  )

							# Forward pass on example inputs (image + text)
							if nlvr_model:
														lowerCamelCase_       :							Optional[int]									=Image.open(requests.get("https://lil.nlp.cornell.edu/nlvr/exs/ex0_0.jpg"				,						stream=a_  ).raw  )
														lowerCamelCase_       :							List[Any]									=Image.open(requests.get("https://lil.nlp.cornell.edu/nlvr/exs/ex0_0.jpg"				,						stream=a_  ).raw  )
														lowerCamelCase_       :							Any									=(
														    '''The left image contains twice the number of dogs as the right image, and at least two dogs in total are'''
														    ''' standing.'''
														)
														lowerCamelCase_       :							Tuple									=processor(a_				,						a_				,						return_tensors="pt"  )
														lowerCamelCase_       :							List[str]									=processor(a_				,						a_				,						return_tensors="pt"  )
														lowerCamelCase_       :							Optional[int]									=model(
														    input_ids=encoding_a.input_ids				,						pixel_values=encoding_a.pixel_values				,						pixel_values_a=encoding_a.pixel_values				,						)
							else:
														lowerCamelCase_       :							Union[str, Any]									=Image.open(requests.get("http://images.cocodataset.org/val2017/000000039769.jpg"				,						stream=a_  ).raw  )
														if mlm_model:
																					lowerCamelCase_       :							Any									='''a bunch of [MASK] laying on a [MASK].'''
														else:
																					lowerCamelCase_       :							Any									='''How many cats are there?'''
														lowerCamelCase_       :							Tuple									=processor(a_				,						a_				,						return_tensors="pt"  )
														lowerCamelCase_       :							Optional[int]									=model(**a_  )

							# Verify outputs
							if mlm_model:
														lowerCamelCase_       :							Any									=torch.Size([1, 11, 30_522]  )
														lowerCamelCase_       :							Dict									=torch.tensor([-12.5061, -12.5123, -12.5174]  )
														assert outputs.logits.shape == expected_shape
														assert torch.allclose(outputs.logits[0, 0, :3]				,						a_				,						atol=1e-4  )

														# verify masked token prediction equals "cats"
														lowerCamelCase_       :							str									=outputs.logits[0, 4, :].argmax(-1  ).item()
														assert tokenizer.decode([predicted_id]  ) == "cats"
							elif vqa_model:
														lowerCamelCase_       :							Any									=torch.Size([1, 3_129]  )
														lowerCamelCase_       :							Tuple									=torch.tensor([-15.9495, -18.1472, -10.3041]  )
														assert torch.allclose(outputs.logits[0, :3]				,						a_				,						atol=1e-4  )
														assert outputs.logits.shape == expected_shape
														assert torch.allclose(outputs.logits[0, 0, :3]				,						a_				,						atol=1e-4  )

														# verify vqa prediction equals "2"
														lowerCamelCase_       :							Optional[Any]									=outputs.logits.argmax(-1  ).item()
														assert model.config.idalabel[predicted_idx] == "2"
							elif nlvr_model:
														lowerCamelCase_       :							List[Any]									=torch.Size([1, 2]  )
														lowerCamelCase_       :							List[str]									=torch.tensor([-2.8721, 2.1291]  )
														assert torch.allclose(outputs.logits[0, :3]				,						a_				,						atol=1e-4  )
														assert outputs.logits.shape == expected_shape

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


if __name__ == "__main__":
						A__       :				Dict 					=					argparse.ArgumentParser()
						# Required parameters
						parser.add_argument(
						    '--checkpoint_url',
						    default='https://github.com/dandelin/ViLT/releases/download/200k/vilt_200k_mlm_itm.ckpt',
						    type=str,
						    help='URL of the checkpoint you\'d like to convert.',
						)
						parser.add_argument(
						    '--pytorch_dump_folder_path', default=None, type=str, help='Path to the output PyTorch model directory.'
						)

						A__       :				Optional[int] 					=					parser.parse_args()
						convert_vilt_checkpoint(args.checkpoint_url, args.pytorch_dump_folder_path)










"""simple docstring"""
import gc
import unittest

import numpy as np
import torch
from torch.backends.cuda import sdp_kernel

from diffusers import (
    CMStochasticIterativeScheduler,
    ConsistencyModelPipeline,
    UNetaDModel,
)
from diffusers.utils import randn_tensor, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_a, require_torch_gpu

from ..pipeline_params import UNCONDITIONAL_IMAGE_GENERATION_BATCH_PARAMS, UNCONDITIONAL_IMAGE_GENERATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()






class 					_SCREAMING_SNAKE_CASE(   A    ,	unittest.TestCase  ):
       SCREAMING_SNAKE_CASE_   :      Optional[Any]							=  ConsistencyModelPipeline
       SCREAMING_SNAKE_CASE_   :      Any							=  UNCONDITIONAL_IMAGE_GENERATION_PARAMS
       SCREAMING_SNAKE_CASE_   :      Dict							=  UNCONDITIONAL_IMAGE_GENERATION_BATCH_PARAMS

       # Override required_optional_params to remove num_images_per_prompt
       SCREAMING_SNAKE_CASE_   :      Optional[Any]							=  frozenset(
           [
               '''num_inference_steps''',
               '''generator''',
               '''latents''',
               '''output_type''',
               '''return_dict''',
               '''callback''',
               '''callback_steps''',
           ]  )




       @property
       def     _UpperCamelCase     (						self							)			->      Tuple:





             """simple docstring"""






             __SCREAMING_SNAKE_CASE   :Tuple						=  UNetaDModel.from_pretrained(
                 '''diffusers/consistency-models-test'''		,subfolder='''test_unet'''		,)
             return unet




       @property
       def     _UpperCamelCase     (						self							)			->      List[Any]:





             """simple docstring"""






             __SCREAMING_SNAKE_CASE   :Optional[int]						=  UNetaDModel.from_pretrained(
                 '''diffusers/consistency-models-test'''		,subfolder='''test_unet_class_cond'''		,)
             return unet




       def     _UpperCamelCase     (						self		,SCREAMING_SNAKE_CASE__=False							)			->      Union[str, Any]:





             """simple docstring"""






             if class_cond:
                   __SCREAMING_SNAKE_CASE   :str						=  self.dummy_cond_unet
             else:
                   __SCREAMING_SNAKE_CASE   :Optional[Any]						=  self.dummy_uncond_unet

             # Default to CM multistep sampler
             __SCREAMING_SNAKE_CASE   :List[str]						=  CMStochasticIterativeScheduler(
                 num_train_timesteps=40		,sigma_min=0.0_0_2		,sigma_max=8_0.0		,)

             __SCREAMING_SNAKE_CASE   :List[str]						=  {
                 '''unet''': unet,
                 '''scheduler''': scheduler,
             }

             return components




       def     _UpperCamelCase     (						self		,SCREAMING_SNAKE_CASE__		,SCREAMING_SNAKE_CASE__=0							)			->      Dict:





             """simple docstring"""






             if str(SCREAMING_SNAKE_CASE__							).startswith('''mps'''							):
                   __SCREAMING_SNAKE_CASE   :Tuple						=  torch.manual_seed(SCREAMING_SNAKE_CASE__							)
             else:
                   __SCREAMING_SNAKE_CASE   :Optional[Any]						=  torch.Generator(device=SCREAMING_SNAKE_CASE__							).manual_seed(SCREAMING_SNAKE_CASE__							)

             __SCREAMING_SNAKE_CASE   :Dict						=  {
                 '''batch_size''': 1,
                 '''num_inference_steps''': None,
                 '''timesteps''': [22, 0],
                 '''generator''': generator,
                 '''output_type''': '''np''',
             }

             return inputs




       def     _UpperCamelCase     (						self							)			->      str:





             """simple docstring"""






             __SCREAMING_SNAKE_CASE   :Union[str, Any]						=  '''cpu'''  # ensure determinism for the device-dependent torch.Generator
             __SCREAMING_SNAKE_CASE   :List[str]						=  self.get_dummy_components()
             __SCREAMING_SNAKE_CASE   :Optional[Any]						=  ConsistencyModelPipeline(**SCREAMING_SNAKE_CASE__							)
             __SCREAMING_SNAKE_CASE   :str						=  pipe.to(SCREAMING_SNAKE_CASE__							)
             pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__							)

             __SCREAMING_SNAKE_CASE   :Tuple						=  self.get_dummy_inputs(SCREAMING_SNAKE_CASE__							)
             __SCREAMING_SNAKE_CASE   :int						=  pipe(**SCREAMING_SNAKE_CASE__							).images
             assert image.shape == (1, 32, 32, 3)

             __SCREAMING_SNAKE_CASE   :List[str]						=  image[0, -3:, -3:, -1]
             __SCREAMING_SNAKE_CASE   :Any						=  np.array([0.3_5_7_2, 0.6_2_7_3, 0.4_0_3_1, 0.3_9_6_1, 0.4_3_2_1, 0.5_7_3_0, 0.5_2_6_6, 0.4_7_8_0, 0.5_0_0_4]							)

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




       def     _UpperCamelCase     (						self							)			->      Dict:





             """simple docstring"""






             __SCREAMING_SNAKE_CASE   :Union[str, Any]						=  '''cpu'''  # ensure determinism for the device-dependent torch.Generator
             __SCREAMING_SNAKE_CASE   :List[Any]						=  self.get_dummy_components(class_cond=SCREAMING_SNAKE_CASE__							)
             __SCREAMING_SNAKE_CASE   :str						=  ConsistencyModelPipeline(**SCREAMING_SNAKE_CASE__							)
             __SCREAMING_SNAKE_CASE   :Optional[int]						=  pipe.to(SCREAMING_SNAKE_CASE__							)
             pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__							)

             __SCREAMING_SNAKE_CASE   :Union[str, Any]						=  self.get_dummy_inputs(SCREAMING_SNAKE_CASE__							)
             __SCREAMING_SNAKE_CASE   :Optional[int]						=  0
             __SCREAMING_SNAKE_CASE   :Optional[int]						=  pipe(**SCREAMING_SNAKE_CASE__							).images
             assert image.shape == (1, 32, 32, 3)

             __SCREAMING_SNAKE_CASE   :Dict						=  image[0, -3:, -3:, -1]
             __SCREAMING_SNAKE_CASE   :List[Any]						=  np.array([0.3_5_7_2, 0.6_2_7_3, 0.4_0_3_1, 0.3_9_6_1, 0.4_3_2_1, 0.5_7_3_0, 0.5_2_6_6, 0.4_7_8_0, 0.5_0_0_4]							)

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




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





             """simple docstring"""






             __SCREAMING_SNAKE_CASE   :List[str]						=  '''cpu'''  # ensure determinism for the device-dependent torch.Generator
             __SCREAMING_SNAKE_CASE   :Tuple						=  self.get_dummy_components()
             __SCREAMING_SNAKE_CASE   :Any						=  ConsistencyModelPipeline(**SCREAMING_SNAKE_CASE__							)
             __SCREAMING_SNAKE_CASE   :List[str]						=  pipe.to(SCREAMING_SNAKE_CASE__							)
             pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__							)

             __SCREAMING_SNAKE_CASE   :str						=  self.get_dummy_inputs(SCREAMING_SNAKE_CASE__							)
             __SCREAMING_SNAKE_CASE   :Dict						=  1
             __SCREAMING_SNAKE_CASE   :List[str]						=  None
             __SCREAMING_SNAKE_CASE   :List[str]						=  pipe(**SCREAMING_SNAKE_CASE__							).images
             assert image.shape == (1, 32, 32, 3)

             __SCREAMING_SNAKE_CASE   :List[str]						=  image[0, -3:, -3:, -1]
             __SCREAMING_SNAKE_CASE   :int						=  np.array([0.5_0_0_4, 0.5_0_0_4, 0.4_9_9_4, 0.5_0_0_8, 0.4_9_7_6, 0.5_0_1_8, 0.4_9_9_0, 0.4_9_8_2, 0.4_9_8_7]							)

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




       def     _UpperCamelCase     (						self							)			->      Any:





             """simple docstring"""






             __SCREAMING_SNAKE_CASE   :str						=  '''cpu'''  # ensure determinism for the device-dependent torch.Generator
             __SCREAMING_SNAKE_CASE   :Any						=  self.get_dummy_components(class_cond=SCREAMING_SNAKE_CASE__							)
             __SCREAMING_SNAKE_CASE   :str						=  ConsistencyModelPipeline(**SCREAMING_SNAKE_CASE__							)
             __SCREAMING_SNAKE_CASE   :Dict						=  pipe.to(SCREAMING_SNAKE_CASE__							)
             pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__							)

             __SCREAMING_SNAKE_CASE   :str						=  self.get_dummy_inputs(SCREAMING_SNAKE_CASE__							)
             __SCREAMING_SNAKE_CASE   :int						=  1
             __SCREAMING_SNAKE_CASE   :Optional[Any]						=  None
             __SCREAMING_SNAKE_CASE   :List[Any]						=  0
             __SCREAMING_SNAKE_CASE   :Any						=  pipe(**SCREAMING_SNAKE_CASE__							).images
             assert image.shape == (1, 32, 32, 3)

             __SCREAMING_SNAKE_CASE   :int						=  image[0, -3:, -3:, -1]
             __SCREAMING_SNAKE_CASE   :Optional[Any]						=  np.array([0.5_0_0_4, 0.5_0_0_4, 0.4_9_9_4, 0.5_0_0_8, 0.4_9_7_6, 0.5_0_1_8, 0.4_9_9_0, 0.4_9_8_2, 0.4_9_8_7]							)

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






@slow
@require_torch_gpu
class 					_SCREAMING_SNAKE_CASE(   unittest.TestCase  ):




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





             """simple docstring"""






             super().tearDown()
             gc.collect()
             torch.cuda.empty_cache()




       def     _UpperCamelCase     (						self		,SCREAMING_SNAKE_CASE__=0		,SCREAMING_SNAKE_CASE__=False		,SCREAMING_SNAKE_CASE__="cpu"		,SCREAMING_SNAKE_CASE__=torch.floataa		,SCREAMING_SNAKE_CASE__=(1, 3, 64, 64)							)			->      Dict:





             """simple docstring"""






             __SCREAMING_SNAKE_CASE   :Tuple						=  torch.manual_seed(SCREAMING_SNAKE_CASE__							)

             __SCREAMING_SNAKE_CASE   :Tuple						=  {
                 '''num_inference_steps''': None,
                 '''timesteps''': [22, 0],
                 '''class_labels''': 0,
                 '''generator''': generator,
                 '''output_type''': '''np''',
             }

             if get_fixed_latents:
                   __SCREAMING_SNAKE_CASE   :int						=  self.get_fixed_latents(seed=SCREAMING_SNAKE_CASE__		,device=SCREAMING_SNAKE_CASE__		,dtype=SCREAMING_SNAKE_CASE__		,shape=SCREAMING_SNAKE_CASE__							)
                   __SCREAMING_SNAKE_CASE   :int						=  latents

             return inputs




       def     _UpperCamelCase     (						self		,SCREAMING_SNAKE_CASE__=0		,SCREAMING_SNAKE_CASE__="cpu"		,SCREAMING_SNAKE_CASE__=torch.floataa		,SCREAMING_SNAKE_CASE__=(1, 3, 64, 64)							)			->      int:





             """simple docstring"""






             if type(SCREAMING_SNAKE_CASE__							) == str:
                   __SCREAMING_SNAKE_CASE   :int						=  torch.device(SCREAMING_SNAKE_CASE__							)
             __SCREAMING_SNAKE_CASE   :Optional[int]						=  torch.Generator(device=SCREAMING_SNAKE_CASE__							).manual_seed(SCREAMING_SNAKE_CASE__							)
             __SCREAMING_SNAKE_CASE   :int						=  randn_tensor(SCREAMING_SNAKE_CASE__		,generator=SCREAMING_SNAKE_CASE__		,device=SCREAMING_SNAKE_CASE__		,dtype=SCREAMING_SNAKE_CASE__							)
             return latents




       def     _UpperCamelCase     (						self							)			->      Any:





             """simple docstring"""






             __SCREAMING_SNAKE_CASE   :List[Any]						=  UNetaDModel.from_pretrained('''diffusers/consistency_models'''		,subfolder='''diffusers_cd_imagenet64_l2'''							)
             __SCREAMING_SNAKE_CASE   :List[str]						=  CMStochasticIterativeScheduler(
                 num_train_timesteps=40		,sigma_min=0.0_0_2		,sigma_max=8_0.0		,)
             __SCREAMING_SNAKE_CASE   :Dict						=  ConsistencyModelPipeline(unet=SCREAMING_SNAKE_CASE__		,scheduler=SCREAMING_SNAKE_CASE__							)
             pipe.to(torch_device=SCREAMING_SNAKE_CASE__							)
             pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__							)

             __SCREAMING_SNAKE_CASE   :int						=  self.get_inputs()
             __SCREAMING_SNAKE_CASE   :List[str]						=  pipe(**SCREAMING_SNAKE_CASE__							).images
             assert image.shape == (1, 64, 64, 3)

             __SCREAMING_SNAKE_CASE   :Union[str, Any]						=  image[0, -3:, -3:, -1]

             __SCREAMING_SNAKE_CASE   :Dict						=  np.array([0.0_8_8_8, 0.0_8_8_1, 0.0_6_6_6, 0.0_4_7_9, 0.0_2_9_2, 0.0_1_9_5, 0.0_2_0_1, 0.0_1_6_3, 0.0_2_5_4]							)

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




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





             """simple docstring"""






             __SCREAMING_SNAKE_CASE   :List[Any]						=  UNetaDModel.from_pretrained('''diffusers/consistency_models'''		,subfolder='''diffusers_cd_imagenet64_l2'''							)
             __SCREAMING_SNAKE_CASE   :List[Any]						=  CMStochasticIterativeScheduler(
                 num_train_timesteps=40		,sigma_min=0.0_0_2		,sigma_max=8_0.0		,)
             __SCREAMING_SNAKE_CASE   :Any						=  ConsistencyModelPipeline(unet=SCREAMING_SNAKE_CASE__		,scheduler=SCREAMING_SNAKE_CASE__							)
             pipe.to(torch_device=SCREAMING_SNAKE_CASE__							)
             pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__							)

             __SCREAMING_SNAKE_CASE   :Optional[int]						=  self.get_inputs()
             __SCREAMING_SNAKE_CASE   :int						=  1
             __SCREAMING_SNAKE_CASE   :int						=  None
             __SCREAMING_SNAKE_CASE   :Union[str, Any]						=  pipe(**SCREAMING_SNAKE_CASE__							).images
             assert image.shape == (1, 64, 64, 3)

             __SCREAMING_SNAKE_CASE   :str						=  image[0, -3:, -3:, -1]

             __SCREAMING_SNAKE_CASE   :List[str]						=  np.array([0.0_3_4_0, 0.0_1_5_2, 0.0_0_6_3, 0.0_2_6_7, 0.0_2_2_1, 0.0_1_0_7, 0.0_4_1_6, 0.0_1_8_6, 0.0_2_1_7]							)

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




       @require_torch_a
       def     _UpperCamelCase     (						self							)			->      str:





             """simple docstring"""






             __SCREAMING_SNAKE_CASE   :List[Any]						=  UNetaDModel.from_pretrained('''diffusers/consistency_models'''		,subfolder='''diffusers_cd_imagenet64_l2'''							)
             __SCREAMING_SNAKE_CASE   :Any						=  CMStochasticIterativeScheduler(
                 num_train_timesteps=40		,sigma_min=0.0_0_2		,sigma_max=8_0.0		,)
             __SCREAMING_SNAKE_CASE   :Any						=  ConsistencyModelPipeline(unet=SCREAMING_SNAKE_CASE__		,scheduler=SCREAMING_SNAKE_CASE__							)
             pipe.to(torch_device=SCREAMING_SNAKE_CASE__		,torch_dtype=torch.floataa							)
             pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__							)

             __SCREAMING_SNAKE_CASE   :str						=  self.get_inputs(get_fixed_latents=SCREAMING_SNAKE_CASE__		,device=SCREAMING_SNAKE_CASE__							)
             # Ensure usage of flash attention in torch 2.0
             with sdp_kernel(enable_flash=SCREAMING_SNAKE_CASE__		,enable_math=SCREAMING_SNAKE_CASE__		,enable_mem_efficient=SCREAMING_SNAKE_CASE__							):
                   __SCREAMING_SNAKE_CASE   :Optional[Any]						=  pipe(**SCREAMING_SNAKE_CASE__							).images
             assert image.shape == (1, 64, 64, 3)

             __SCREAMING_SNAKE_CASE   :List[str]						=  image[0, -3:, -3:, -1]

             __SCREAMING_SNAKE_CASE   :List[Any]						=  np.array([0.1_8_7_5, 0.1_4_2_8, 0.1_2_8_9, 0.2_1_5_1, 0.2_0_9_2, 0.1_4_7_7, 0.1_8_7_7, 0.1_6_4_1, 0.1_3_5_3]							)

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




       @require_torch_a
       def     _UpperCamelCase     (						self							)			->      List[Any]:





             """simple docstring"""






             __SCREAMING_SNAKE_CASE   :str						=  UNetaDModel.from_pretrained('''diffusers/consistency_models'''		,subfolder='''diffusers_cd_imagenet64_l2'''							)
             __SCREAMING_SNAKE_CASE   :Dict						=  CMStochasticIterativeScheduler(
                 num_train_timesteps=40		,sigma_min=0.0_0_2		,sigma_max=8_0.0		,)
             __SCREAMING_SNAKE_CASE   :int						=  ConsistencyModelPipeline(unet=SCREAMING_SNAKE_CASE__		,scheduler=SCREAMING_SNAKE_CASE__							)
             pipe.to(torch_device=SCREAMING_SNAKE_CASE__		,torch_dtype=torch.floataa							)
             pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__							)

             __SCREAMING_SNAKE_CASE   :Dict						=  self.get_inputs(get_fixed_latents=SCREAMING_SNAKE_CASE__		,device=SCREAMING_SNAKE_CASE__							)
             __SCREAMING_SNAKE_CASE   :int						=  1
             __SCREAMING_SNAKE_CASE   :int						=  None
             # Ensure usage of flash attention in torch 2.0
             with sdp_kernel(enable_flash=SCREAMING_SNAKE_CASE__		,enable_math=SCREAMING_SNAKE_CASE__		,enable_mem_efficient=SCREAMING_SNAKE_CASE__							):
                   __SCREAMING_SNAKE_CASE   :str						=  pipe(**SCREAMING_SNAKE_CASE__							).images
             assert image.shape == (1, 64, 64, 3)

             __SCREAMING_SNAKE_CASE   :str						=  image[0, -3:, -3:, -1]

             __SCREAMING_SNAKE_CASE   :Optional[int]						=  np.array([0.1_6_6_3, 0.1_9_4_8, 0.2_2_7_5, 0.1_6_8_0, 0.1_2_0_4, 0.1_2_4_5, 0.1_8_5_8, 0.1_3_3_8, 0.2_0_9_5]							)

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