import unittest

from transformers import TrOCRConfig
from transformers.testing_utils import is_torch_available, require_torch, torch_device

from ...generation.test_utils import GenerationTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
				import torch

				from transformers.models.trocr.modeling_trocr import TrOCRDecoder, TrOCRForCausalLM



@require_torch
class        lowercase_   :



							def __init__(						self				,    lowercase_				,    lowercase_=99				,    lowercase_=13				,    lowercase_=16				,    lowercase_=7				,    lowercase_=True				,    lowercase_=True				,    lowercase_=True				,    lowercase_=False				,    lowercase_=True				,    lowercase_=2				,    lowercase_=32				,    lowercase_=4				,    lowercase_=4				,    lowercase_=30				,    lowercase_=0				,    lowercase_=1				,    lowercase_=2				,    lowercase_=None				,    ):
													_snake_case						:					Tuple    =   parent
													_snake_case						:					Tuple    =   batch_size
													_snake_case						:					Any    =   decoder_seq_length
													# For common tests
													_snake_case						:					int    =   self.decoder_seq_length
													_snake_case						:					int    =   is_training
													_snake_case						:					Dict    =   use_attention_mask
													_snake_case						:					int    =   use_labels

													_snake_case						:					List[str]    =   vocab_size
													_snake_case						:					Optional[int]    =   d_model
													_snake_case						:					Optional[Any]    =   d_model
													_snake_case						:					List[str]    =   decoder_layers
													_snake_case						:					Tuple    =   decoder_layers
													_snake_case						:					List[str]    =   decoder_ffn_dim
													_snake_case						:					List[Any]    =   decoder_attention_heads
													_snake_case						:					Tuple    =   decoder_attention_heads
													_snake_case						:					Dict    =   eos_token_id
													_snake_case						:					Dict    =   bos_token_id
													_snake_case						:					Any    =   pad_token_id
													_snake_case						:					Union[str, Any]    =   decoder_start_token_id
													_snake_case						:					List[Any]    =   use_cache
													_snake_case						:					Optional[int]    =   max_position_embeddings

													_snake_case						:					Optional[int]    =   None
													_snake_case						:					Tuple    =   decoder_seq_length
													_snake_case						:					List[str]    =   2
													_snake_case						:					int    =   1



							def 	UpperCamelCase	(						self    ):
													_snake_case						:					Any    =   ids_tensor([self.batch_size, self.decoder_seq_length]				,    self.vocab_size    )

													_snake_case						:					Union[str, Any]    =   None
													if self.use_attention_mask:
																			_snake_case						:					Any    =   ids_tensor([self.batch_size, self.decoder_seq_length]				,    vocab_size=2    )

													_snake_case						:					str    =   None
													if self.use_labels:
																			_snake_case						:					Optional[int]    =   ids_tensor([self.batch_size, self.decoder_seq_length]				,    self.vocab_size    )

													_snake_case						:					int    =   TrOCRConfig(
													    vocab_size=self.vocab_size				,    d_model=self.d_model				,    decoder_layers=self.decoder_layers				,    decoder_ffn_dim=self.decoder_ffn_dim				,    decoder_attention_heads=self.decoder_attention_heads				,    eos_token_id=self.eos_token_id				,    bos_token_id=self.bos_token_id				,    use_cache=self.use_cache				,    pad_token_id=self.pad_token_id				,    decoder_start_token_id=self.decoder_start_token_id				,    max_position_embeddings=self.max_position_embeddings				,    )

													return (config, input_ids, attention_mask, lm_labels)



							def 	UpperCamelCase	(						self				,    lowercase_				,    lowercase_				,    lowercase_				,    lowercase_				,    ):
													_snake_case						:					Any    =   True
													_snake_case						:					int    =   TrOCRDecoder(config=__lowerCamelCase    ).to(__lowerCamelCase    ).eval()
													_snake_case						:					Any    =   input_ids[:2]

													input_ids[input_ids == 0] += 1
													# first forward pass
													_snake_case						:					Any    =   model(__lowerCamelCase				,    use_cache=__lowerCamelCase    )
													_snake_case						:					List[Any]    =   model(__lowerCamelCase    )
													_snake_case						:					List[Any]    =   model(__lowerCamelCase				,    use_cache=__lowerCamelCase    )

													self.parent.assertTrue(len(__lowerCamelCase    ) == len(__lowerCamelCase    )    )
													self.parent.assertTrue(len(__lowerCamelCase    ) == len(__lowerCamelCase    ) + 1    )

													_snake_case						:					Any    =   outputs["past_key_values"]

													# create hypothetical next token and extent to next_input_ids
													_snake_case						:					int    =   ids_tensor((2, 1)				,    config.vocab_size - 1    ) + 1

													# append to next input_ids and
													_snake_case						:					str    =   torch.cat([input_ids, next_tokens]				,    dim=-1    )

													_snake_case						:					List[Any]    =   model(__lowerCamelCase    )["last_hidden_state"]
													_snake_case						:					Optional[Any]    =   model(__lowerCamelCase				,    past_key_values=__lowerCamelCase    )["last_hidden_state"]

													# select random slice
													_snake_case						:					List[Any]    =   ids_tensor((1,)				,    output_from_past.shape[-1]    ).item()
													_snake_case						:					List[str]    =   output_from_no_past[:, next_input_ids.shape[-1] - 1, random_slice_idx].detach()
													_snake_case						:					int    =   output_from_past[:, 0, random_slice_idx].detach()

													# test that outputs are equal for slice
													assert torch.allclose(__lowerCamelCase				,    __lowerCamelCase				,    atol=1e-3    )



							def 	UpperCamelCase	(						self    ):
													_snake_case						:					str    =   self.prepare_config_and_inputs()
													_snake_case    ,_snake_case    ,_snake_case    ,_snake_case						:					Any    =   config_and_inputs

													_snake_case						:					int    =   {"input_ids": input_ids, "attention_mask": attention_mask}
													return config, inputs_dict




@require_torch
class        lowercase_   (					UpperCamelCase__						,			UpperCamelCase__						,			UpperCamelCase__						,			unittest.TestCase		):
							_lowerCamelCase						=		(TrOCRDecoder, TrOCRForCausalLM) if is_torch_available() else ()
							_lowerCamelCase						=		(TrOCRForCausalLM,) if is_torch_available() else ()
							_lowerCamelCase						=		{'text-generation': TrOCRForCausalLM} if is_torch_available() else {}
							_lowerCamelCase						=		True
							_lowerCamelCase						=		False



							def 	UpperCamelCase	(						self    ):
													_snake_case						:					Union[str, Any]    =   TrOCRStandaloneDecoderModelTester(self				,    is_training=__lowerCamelCase    )
													_snake_case						:					Optional[Any]    =   ConfigTester(self				,    config_class=__lowerCamelCase    )



							def 	UpperCamelCase	(						self    ):
													pass



							def 	UpperCamelCase	(						self    ):
													pass



							def 	UpperCamelCase	(						self    ):
													pass



							def 	UpperCamelCase	(						self    ):
													self.config_tester.run_common_tests()



							def 	UpperCamelCase	(						self    ):
													_snake_case						:					Optional[Any]    =   self.model_tester.prepare_config_and_inputs()
													self.model_tester.create_and_check_decoder_model_past(*__lowerCamelCase    )



							def 	UpperCamelCase	(						self    ):
													return



							@unittest.skip("The model doesn\'t support left padding"    )  # and it's not used enough to be worth fixing :)
							def 	UpperCamelCase	(						self    ):
													pass
import tempfile

import numpy as np
import torch
from transformers import AutoTokenizer, TaEncoderModel

from diffusers import DDPMScheduler, UNetaDConditionModel
from diffusers.models.attention_processor import AttnAddedKVProcessor
from diffusers.pipelines.deepfloyd_if import IFWatermarker
from diffusers.utils.testing_utils import torch_device

from ..test_pipelines_common import to_np



class        lowercase_   :



							def 	UpperCamelCase	(						self    ):
													torch.manual_seed(0    )
													_snake_case						:					Tuple    =   TaEncoderModel.from_pretrained("hf-internal-testing/tiny-random-t5"    )

													torch.manual_seed(0    )
													_snake_case						:					List[Any]    =   AutoTokenizer.from_pretrained("hf-internal-testing/tiny-random-t5"    )

													torch.manual_seed(0    )
													_snake_case						:					Optional[int]    =   UNetaDConditionModel(
													    sample_size=32				,    layers_per_block=1				,    block_out_channels=[32, 64]				,    down_block_types=[
													        "ResnetDownsampleBlock2D",
													        "SimpleCrossAttnDownBlock2D",
													    ]				,    mid_block_type="UNetMidBlock2DSimpleCrossAttn"				,    up_block_types=["SimpleCrossAttnUpBlock2D", "ResnetUpsampleBlock2D"]				,    in_channels=3				,    out_channels=6				,    cross_attention_dim=32				,    encoder_hid_dim=32				,    attention_head_dim=8				,    addition_embed_type="text"				,    addition_embed_type_num_heads=2				,    cross_attention_norm="group_norm"				,    resnet_time_scale_shift="scale_shift"				,    act_fn="gelu"				,    )
													unet.set_attn_processor(AttnAddedKVProcessor()    )  # For reproducibility tests

													torch.manual_seed(0    )
													_snake_case						:					str    =   DDPMScheduler(
													    num_train_timesteps=1_000				,    beta_schedule="squaredcos_cap_v2"				,    beta_start=0.0_001				,    beta_end=0.02				,    thresholding=lowercase_				,    dynamic_thresholding_ratio=0.95				,    sample_max_value=1.0				,    prediction_type="epsilon"				,    variance_type="learned_range"				,    )

													torch.manual_seed(0    )
													_snake_case						:					int    =   IFWatermarker()

													return {
													    "text_encoder": text_encoder,
													    "tokenizer": tokenizer,
													    "unet": unet,
													    "scheduler": scheduler,
													    "watermarker": watermarker,
													    "safety_checker": None,
													    "feature_extractor": None,
													}



							def 	UpperCamelCase	(						self    ):
													torch.manual_seed(0    )
													_snake_case						:					List[str]    =   TaEncoderModel.from_pretrained("hf-internal-testing/tiny-random-t5"    )

													torch.manual_seed(0    )
													_snake_case						:					Tuple    =   AutoTokenizer.from_pretrained("hf-internal-testing/tiny-random-t5"    )

													torch.manual_seed(0    )
													_snake_case						:					Any    =   UNetaDConditionModel(
													    sample_size=32				,    layers_per_block=[1, 2]				,    block_out_channels=[32, 64]				,    down_block_types=[
													        "ResnetDownsampleBlock2D",
													        "SimpleCrossAttnDownBlock2D",
													    ]				,    mid_block_type="UNetMidBlock2DSimpleCrossAttn"				,    up_block_types=["SimpleCrossAttnUpBlock2D", "ResnetUpsampleBlock2D"]				,    in_channels=6				,    out_channels=6				,    cross_attention_dim=32				,    encoder_hid_dim=32				,    attention_head_dim=8				,    addition_embed_type="text"				,    addition_embed_type_num_heads=2				,    cross_attention_norm="group_norm"				,    resnet_time_scale_shift="scale_shift"				,    act_fn="gelu"				,    class_embed_type="timestep"				,    mid_block_scale_factor=1.414				,    time_embedding_act_fn="gelu"				,    time_embedding_dim=32				,    )
													unet.set_attn_processor(AttnAddedKVProcessor()    )  # For reproducibility tests

													torch.manual_seed(0    )
													_snake_case						:					Union[str, Any]    =   DDPMScheduler(
													    num_train_timesteps=1_000				,    beta_schedule="squaredcos_cap_v2"				,    beta_start=0.0_001				,    beta_end=0.02				,    thresholding=lowercase_				,    dynamic_thresholding_ratio=0.95				,    sample_max_value=1.0				,    prediction_type="epsilon"				,    variance_type="learned_range"				,    )

													torch.manual_seed(0    )
													_snake_case						:					List[str]    =   DDPMScheduler(
													    num_train_timesteps=1_000				,    beta_schedule="squaredcos_cap_v2"				,    beta_start=0.0_001				,    beta_end=0.02				,    )

													torch.manual_seed(0    )
													_snake_case						:					Dict    =   IFWatermarker()

													return {
													    "text_encoder": text_encoder,
													    "tokenizer": tokenizer,
													    "unet": unet,
													    "scheduler": scheduler,
													    "image_noising_scheduler": image_noising_scheduler,
													    "watermarker": watermarker,
													    "safety_checker": None,
													    "feature_extractor": None,
													}



							def 	UpperCamelCase	(						self    ):
													_snake_case						:					List[Any]    =   self.get_dummy_components()
													_snake_case						:					List[Any]    =   self.pipeline_class(**lowercase_    )
													pipe.to(lowercase_    )
													pipe.set_progress_bar_config(disable=lowercase_    )

													_snake_case						:					Union[str, Any]    =   self.get_dummy_inputs(lowercase_    )

													_snake_case						:					Union[str, Any]    =   inputs["prompt"]
													_snake_case						:					Dict    =   inputs["generator"]
													_snake_case						:					Any    =   inputs["num_inference_steps"]
													_snake_case						:					Union[str, Any]    =   inputs["output_type"]

													if "image" in inputs:
																			_snake_case						:					int    =   inputs["image"]
													else:
																			_snake_case						:					Union[str, Any]    =   None

													if "mask_image" in inputs:
																			_snake_case						:					int    =   inputs["mask_image"]
													else:
																			_snake_case						:					List[str]    =   None

													if "original_image" in inputs:
																			_snake_case						:					Tuple    =   inputs["original_image"]
													else:
																			_snake_case						:					Any    =   None

													_snake_case    ,_snake_case						:					Optional[int]    =   pipe.encode_prompt(lowercase_    )

													# inputs with prompt converted to embeddings
													_snake_case						:					Dict    =   {
													    "prompt_embeds": prompt_embeds,
													    "negative_prompt_embeds": negative_prompt_embeds,
													    "generator": generator,
													    "num_inference_steps": num_inference_steps,
													    "output_type": output_type,
													}

													if image is not None:
																			_snake_case						:					int    =   image

													if mask_image is not None:
																			_snake_case						:					int    =   mask_image

													if original_image is not None:
																			_snake_case						:					Any    =   original_image

													# set all optional components to None
													for optional_component in pipe._optional_components:
																			setattr(lowercase_				,    lowercase_				,    lowercase_    )

													_snake_case						:					Optional[int]    =   pipe(**lowercase_    )[0]

													with tempfile.TemporaryDirectory() as tmpdir:
																			pipe.save_pretrained(lowercase_    )
																			_snake_case						:					Any    =   self.pipeline_class.from_pretrained(lowercase_    )
																			pipe_loaded.to(lowercase_    )
																			pipe_loaded.set_progress_bar_config(disable=lowercase_    )

													pipe_loaded.unet.set_attn_processor(AttnAddedKVProcessor()    )  # For reproducibility tests

													for optional_component in pipe._optional_components:
																			self.assertTrue(
																			    getattr(lowercase_				,    lowercase_    ) is None				,    f"""`{optional_component}` did not stay set to None after loading."""				,    )

													_snake_case						:					int    =   self.get_dummy_inputs(lowercase_    )

													_snake_case						:					Optional[int]    =   inputs["generator"]
													_snake_case						:					List[Any]    =   inputs["num_inference_steps"]
													_snake_case						:					Tuple    =   inputs["output_type"]

													# inputs with prompt converted to embeddings
													_snake_case						:					int    =   {
													    "prompt_embeds": prompt_embeds,
													    "negative_prompt_embeds": negative_prompt_embeds,
													    "generator": generator,
													    "num_inference_steps": num_inference_steps,
													    "output_type": output_type,
													}

													if image is not None:
																			_snake_case						:					int    =   image

													if mask_image is not None:
																			_snake_case						:					str    =   mask_image

													if original_image is not None:
																			_snake_case						:					int    =   original_image

													_snake_case						:					Optional[Any]    =   pipe_loaded(**lowercase_    )[0]

													_snake_case						:					Dict    =   np.abs(to_np(lowercase_    ) - to_np(lowercase_    )    ).max()
													self.assertLess(lowercase_				,    1e-4    )



							def 	UpperCamelCase	(						self    ):
													_snake_case						:					Tuple    =   self.get_dummy_components()
													_snake_case						:					Any    =   self.pipeline_class(**lowercase_    )
													pipe.to(lowercase_    )
													pipe.set_progress_bar_config(disable=lowercase_    )

													_snake_case						:					int    =   self.get_dummy_inputs(lowercase_    )
													_snake_case						:					List[str]    =   pipe(**lowercase_    )[0]

													with tempfile.TemporaryDirectory() as tmpdir:
																			pipe.save_pretrained(lowercase_    )
																			_snake_case						:					List[Any]    =   self.pipeline_class.from_pretrained(lowercase_    )
																			pipe_loaded.to(lowercase_    )
																			pipe_loaded.set_progress_bar_config(disable=lowercase_    )

													pipe_loaded.unet.set_attn_processor(AttnAddedKVProcessor()    )  # For reproducibility tests

													_snake_case						:					Optional[Any]    =   self.get_dummy_inputs(lowercase_    )
													_snake_case						:					int    =   pipe_loaded(**lowercase_    )[0]

													_snake_case						:					Tuple    =   np.abs(to_np(lowercase_    ) - to_np(lowercase_    )    ).max()
													self.assertLess(lowercase_				,    1e-4    )