






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

import numpy as np
import torch
from transformers import (
    CLIPImageProcessor,
    CLIPTextConfig,
    CLIPTextModelWithProjection,
    CLIPTokenizer,
    CLIPVisionConfig,
    CLIPVisionModelWithProjection,
)

from diffusers import (
    DiffusionPipeline,
    UnCLIPImageVariationPipeline,
    UnCLIPScheduler,
    UNetaDConditionModel,
    UNetaDModel,
)
from diffusers.pipelines.unclip.text_proj import UnCLIPTextProjModel
from diffusers.utils import floats_tensor, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, load_image, require_torch_gpu, skip_mps

from ..pipeline_params import IMAGE_VARIATION_BATCH_PARAMS, IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()



class        UpperCamelCase__ (				a_							,	unittest.TestCase):







    """simple docstring"""
    __UpperCAmelCase           = UnCLIPImageVariationPipeline
    __UpperCAmelCase           = IMAGE_VARIATION_PARAMS - {"""height""", """width""", """guidance_scale"""}
    __UpperCAmelCase           = IMAGE_VARIATION_BATCH_PARAMS

    __UpperCAmelCase           = [
        """generator""",
        """return_dict""",
        """decoder_num_inference_steps""",
        """super_res_num_inference_steps""",
    ]
    __UpperCAmelCase           = False







    @property
    def 				a__			(		self :	str ):





           '''simple docstring'''

           return 3_2







    @property
    def 				a__			(		self :	Dict ):





           '''simple docstring'''

           return 3_2







    @property
    def 				a__			(		self :	Optional[int] ):





           '''simple docstring'''

           return self.time_input_dim







    @property
    def 				a__			(		self :	str ):





           '''simple docstring'''

           return self.time_input_dim * 4







    @property
    def 				a__			(		self :	Union[str, Any] ):





           '''simple docstring'''

           return 1_0_0







    @property
    def 				a__			(		self :	Tuple ):





           '''simple docstring'''

           __magic_name__      				=    CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip' )
           return tokenizer







    @property
    def 				a__			(		self :	Dict ):





           '''simple docstring'''

           torch.manual_seed(0 )
           __magic_name__      				=    CLIPTextConfig(
               bos_token_id=0				,    eos_token_id=2				,    hidden_size=self.text_embedder_hidden_size				,    projection_dim=self.text_embedder_hidden_size				,    intermediate_size=3_7				,    layer_norm_eps=1e-0_5				,    num_attention_heads=4				,    num_hidden_layers=5				,    pad_token_id=1				,    vocab_size=1_0_0_0				,    )
           return CLIPTextModelWithProjection(UpperCamelCase_ )







    @property
    def 				a__			(		self :	int ):





           '''simple docstring'''

           torch.manual_seed(0 )
           __magic_name__      				=    CLIPVisionConfig(
               hidden_size=self.text_embedder_hidden_size				,    projection_dim=self.text_embedder_hidden_size				,    num_hidden_layers=5				,    num_attention_heads=4				,    image_size=3_2				,    intermediate_size=3_7				,    patch_size=1				,    )
           return CLIPVisionModelWithProjection(UpperCamelCase_ )







    @property
    def 				a__			(		self :	List[Any] ):





           '''simple docstring'''

           torch.manual_seed(0 )

           __magic_name__      				=    {
               'clip_embeddings_dim': self.text_embedder_hidden_size,
               'time_embed_dim': self.time_embed_dim,
               'cross_attention_dim': self.cross_attention_dim,
           }

           __magic_name__      				=    UnCLIPTextProjModel(**UpperCamelCase_ )
           return model







    @property
    def 				a__			(		self :	Union[str, Any] ):





           '''simple docstring'''

           torch.manual_seed(0 )

           __magic_name__      				=    {
               'sample_size': 3_2,
               # RGB in channels
               'in_channels': 3,
               # Out channels is double in channels because predicts mean and variance
               'out_channels': 6,
               'down_block_types': ('ResnetDownsampleBlock2D', 'SimpleCrossAttnDownBlock2D'),
               'up_block_types': ('SimpleCrossAttnUpBlock2D', 'ResnetUpsampleBlock2D'),
               'mid_block_type': 'UNetMidBlock2DSimpleCrossAttn',
               'block_out_channels': (self.block_out_channels_a, self.block_out_channels_a * 2),
               'layers_per_block': 1,
               'cross_attention_dim': self.cross_attention_dim,
               'attention_head_dim': 4,
               'resnet_time_scale_shift': 'scale_shift',
               'class_embed_type': 'identity',
           }

           __magic_name__      				=    UNetaDConditionModel(**UpperCamelCase_ )
           return model







    @property
    def 				a__			(		self :	Any ):





           '''simple docstring'''

           return {
               "sample_size": 6_4,
               "layers_per_block": 1,
               "down_block_types": ("ResnetDownsampleBlock2D", "ResnetDownsampleBlock2D"),
               "up_block_types": ("ResnetUpsampleBlock2D", "ResnetUpsampleBlock2D"),
               "block_out_channels": (self.block_out_channels_a, self.block_out_channels_a * 2),
               "in_channels": 6,
               "out_channels": 3,
           }







    @property
    def 				a__			(		self :	str ):





           '''simple docstring'''

           torch.manual_seed(0 )

           __magic_name__      				=    UNetaDModel(**self.dummy_super_res_kwargs )
           return model







    @property
    def 				a__			(		self :	Union[str, Any] ):





           '''simple docstring'''

           torch.manual_seed(1 )

           __magic_name__      				=    UNetaDModel(**self.dummy_super_res_kwargs )
           return model







    def 				a__			(		self :	Dict ):





           '''simple docstring'''

           __magic_name__      				=    self.dummy_decoder
           __magic_name__      				=    self.dummy_text_proj
           __magic_name__      				=    self.dummy_text_encoder
           __magic_name__      				=    self.dummy_tokenizer
           __magic_name__      				=    self.dummy_super_res_first
           __magic_name__      				=    self.dummy_super_res_last

           __magic_name__      				=    UnCLIPScheduler(
               variance_type='learned_range'				,    prediction_type='epsilon'				,    num_train_timesteps=1_0_0_0				,    )

           __magic_name__      				=    UnCLIPScheduler(
               variance_type='fixed_small_log'				,    prediction_type='epsilon'				,    num_train_timesteps=1_0_0_0				,    )

           __magic_name__      				=    CLIPImageProcessor(crop_size=3_2				,    size=3_2 )

           __magic_name__      				=    self.dummy_image_encoder

           return {
               "decoder": decoder,
               "text_encoder": text_encoder,
               "tokenizer": tokenizer,
               "text_proj": text_proj,
               "feature_extractor": feature_extractor,
               "image_encoder": image_encoder,
               "super_res_first": super_res_first,
               "super_res_last": super_res_last,
               "decoder_scheduler": decoder_scheduler,
               "super_res_scheduler": super_res_scheduler,
           }







    def 				a__			(		self :	Union[str, Any]				,    UpperCamelCase_ :	Dict				,    UpperCamelCase_ :	Optional[int]=0				,    UpperCamelCase_ :	Optional[Any]=True ):





           '''simple docstring'''

           __magic_name__      				=    floats_tensor((1, 3, 3_2, 3_2)				,    rng=random.Random(UpperCamelCase_ ) ).to(UpperCamelCase_ )
           if str(UpperCamelCase_ ).startswith('mps' ):
                  __magic_name__      				=    torch.manual_seed(UpperCamelCase_ )
           else:
                  __magic_name__      				=    torch.Generator(device=UpperCamelCase_ ).manual_seed(UpperCamelCase_ )

           if pil_image:
                  __magic_name__      				=    input_image * 0.5 + 0.5
                  __magic_name__      				=    input_image.clamp(0				,    1 )
                  __magic_name__      				=    input_image.cpu().permute(0				,    2				,    3				,    1 ).float().numpy()
                  __magic_name__      				=    DiffusionPipeline.numpy_to_pil(UpperCamelCase_ )[0]

           return {
               "image": input_image,
               "generator": generator,
               "decoder_num_inference_steps": 2,
               "super_res_num_inference_steps": 2,
               "output_type": "np",
           }







    def 				a__			(		self :	List[Any] ):





           '''simple docstring'''

           __magic_name__      				=    'cpu'

           __magic_name__      				=    self.get_dummy_components()

           __magic_name__      				=    self.pipeline_class(**UpperCamelCase_ )
           __magic_name__      				=    pipe.to(UpperCamelCase_ )

           pipe.set_progress_bar_config(disable=UpperCamelCase_ )

           __magic_name__      				=    self.get_dummy_inputs(UpperCamelCase_				,    pil_image=UpperCamelCase_ )

           __magic_name__      				=    pipe(**UpperCamelCase_ )
           __magic_name__      				=    output.images

           __magic_name__      				=    self.get_dummy_inputs(UpperCamelCase_				,    pil_image=UpperCamelCase_ )

           __magic_name__      				=    pipe(
               **UpperCamelCase_				,    return_dict=UpperCamelCase_				,    )[0]

           __magic_name__      				=    image[0, -3:, -3:, -1]
           __magic_name__      				=    image_from_tuple[0, -3:, -3:, -1]

           assert image.shape == (1, 6_4, 6_4, 3)

           __magic_name__      				=    np.array(
               [
                   0.9_997,
                   0.0_002,
                   0.9_997,
                   0.9_997,
                   0.9_969,
                   0.0_023,
                   0.9_997,
                   0.9_969,
                   0.9_970,
               ] )

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







    def 				a__			(		self :	Union[str, Any] ):





           '''simple docstring'''

           __magic_name__      				=    'cpu'

           __magic_name__      				=    self.get_dummy_components()

           __magic_name__      				=    self.pipeline_class(**UpperCamelCase_ )
           __magic_name__      				=    pipe.to(UpperCamelCase_ )

           pipe.set_progress_bar_config(disable=UpperCamelCase_ )

           __magic_name__      				=    self.get_dummy_inputs(UpperCamelCase_				,    pil_image=UpperCamelCase_ )

           __magic_name__      				=    pipe(**UpperCamelCase_ )
           __magic_name__      				=    output.images

           __magic_name__      				=    self.get_dummy_inputs(UpperCamelCase_				,    pil_image=UpperCamelCase_ )

           __magic_name__      				=    pipe(
               **UpperCamelCase_				,    return_dict=UpperCamelCase_				,    )[0]

           __magic_name__      				=    image[0, -3:, -3:, -1]
           __magic_name__      				=    image_from_tuple[0, -3:, -3:, -1]

           assert image.shape == (1, 6_4, 6_4, 3)

           __magic_name__      				=    np.array([0.9_997, 0.0_003, 0.9_997, 0.9_997, 0.9_970, 0.0_024, 0.9_997, 0.9_971, 0.9_971] )

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







    def 				a__			(		self :	Any ):





           '''simple docstring'''

           __magic_name__      				=    'cpu'

           __magic_name__      				=    self.get_dummy_components()

           __magic_name__      				=    self.pipeline_class(**UpperCamelCase_ )
           __magic_name__      				=    pipe.to(UpperCamelCase_ )

           pipe.set_progress_bar_config(disable=UpperCamelCase_ )

           __magic_name__      				=    self.get_dummy_inputs(UpperCamelCase_				,    pil_image=UpperCamelCase_ )
           __magic_name__      				=    [
               pipeline_inputs['image'],
               pipeline_inputs['image'],
           ]

           __magic_name__      				=    pipe(**UpperCamelCase_ )
           __magic_name__      				=    output.images

           __magic_name__      				=    self.get_dummy_inputs(UpperCamelCase_				,    pil_image=UpperCamelCase_ )
           __magic_name__      				=    [
               tuple_pipeline_inputs['image'],
               tuple_pipeline_inputs['image'],
           ]

           __magic_name__      				=    pipe(
               **UpperCamelCase_				,    return_dict=UpperCamelCase_				,    )[0]

           __magic_name__      				=    image[0, -3:, -3:, -1]
           __magic_name__      				=    image_from_tuple[0, -3:, -3:, -1]

           assert image.shape == (2, 6_4, 6_4, 3)

           __magic_name__      				=    np.array(
               [
                   0.9_997,
                   0.9_989,
                   0.0_008,
                   0.0_021,
                   0.9_960,
                   0.0_018,
                   0.0_014,
                   0.0_002,
                   0.9_933,
               ] )

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







    def 				a__			(		self :	Optional[int] ):





           '''simple docstring'''

           __magic_name__      				=    torch.device('cpu' )



           class        UpperCamelCase__ :







               """simple docstring"""
               __UpperCAmelCase           = 1
           __magic_name__      				=    self.get_dummy_components()

           __magic_name__      				=    self.pipeline_class(**UpperCamelCase_ )
           __magic_name__      				=    pipe.to(UpperCamelCase_ )

           pipe.set_progress_bar_config(disable=UpperCamelCase_ )

           __magic_name__      				=    torch.Generator(device=UpperCamelCase_ ).manual_seed(0 )
           __magic_name__      				=    pipe.decoder.dtype
           __magic_name__      				=    1

           __magic_name__      				=    (
               batch_size,
               pipe.decoder.config.in_channels,
               pipe.decoder.config.sample_size,
               pipe.decoder.config.sample_size,
           )
           __magic_name__      				=    pipe.prepare_latents(
               UpperCamelCase_				,    dtype=UpperCamelCase_				,    device=UpperCamelCase_				,    generator=UpperCamelCase_				,    latents=UpperCamelCase_				,    scheduler=DummyScheduler() )

           __magic_name__      				=    (
               batch_size,
               pipe.super_res_first.config.in_channels // 2,
               pipe.super_res_first.config.sample_size,
               pipe.super_res_first.config.sample_size,
           )
           __magic_name__      				=    pipe.prepare_latents(
               UpperCamelCase_				,    dtype=UpperCamelCase_				,    device=UpperCamelCase_				,    generator=UpperCamelCase_				,    latents=UpperCamelCase_				,    scheduler=DummyScheduler() )

           __magic_name__      				=    self.get_dummy_inputs(UpperCamelCase_				,    pil_image=UpperCamelCase_ )

           __magic_name__      				=    pipe(
               **UpperCamelCase_				,    decoder_latents=UpperCamelCase_				,    super_res_latents=UpperCamelCase_ ).images

           __magic_name__      				=    self.get_dummy_inputs(UpperCamelCase_				,    pil_image=UpperCamelCase_ )
           # Don't pass image, instead pass embedding
           __magic_name__      				=    pipeline_inputs.pop('image' )
           __magic_name__      				=    pipe.image_encoder(UpperCamelCase_ ).image_embeds

           __magic_name__      				=    pipe(
               **UpperCamelCase_				,    decoder_latents=UpperCamelCase_				,    super_res_latents=UpperCamelCase_				,    image_embeddings=UpperCamelCase_				,    ).images

           # make sure passing text embeddings manually is identical
           assert np.abs(img_out_a - img_out_a ).max() < 1e-4







    @skip_mps
    def 				a__			(		self :	Dict ):





           '''simple docstring'''

           __magic_name__      				=    torch_device == 'cpu'

           # Check is relaxed because there is not a torch 2.0 sliced attention added kv processor
           __magic_name__      				=    1e-2

           self._test_attention_slicing_forward_pass(
               test_max_difference=UpperCamelCase_				,    expected_max_diff=UpperCamelCase_ )







    @skip_mps
    def 				a__			(		self :	Tuple ):





           '''simple docstring'''

           __magic_name__      				=    torch_device == 'cpu'
           __magic_name__      				=    True
           __magic_name__      				=    [
               'decoder_num_inference_steps',
               'super_res_num_inference_steps',
           ]

           self._test_inference_batch_single_identical(
               test_max_difference=UpperCamelCase_				,    relax_max_difference=UpperCamelCase_				,    additional_params_copy_to_batched_inputs=UpperCamelCase_				,    )







    def 				a__			(		self :	Dict ):





           '''simple docstring'''

           __magic_name__      				=    [
               'decoder_num_inference_steps',
               'super_res_num_inference_steps',
           ]

           if torch_device == "mps":
                  # TODO: MPS errors with larger batch sizes
                  __magic_name__      				=    [2, 3]
                  self._test_inference_batch_consistent(
                      batch_sizes=UpperCamelCase_				,    additional_params_copy_to_batched_inputs=UpperCamelCase_				,    )
           else:
                  self._test_inference_batch_consistent(
                      additional_params_copy_to_batched_inputs=UpperCamelCase_ )







    @skip_mps
    def 				a__			(		self :	List[Any] ):





           '''simple docstring'''

           return super().test_dict_tuple_outputs_equivalent()







    @skip_mps
    def 				a__			(		self :	int ):





           '''simple docstring'''

           return super().test_save_load_local()







    @skip_mps
    def 				a__			(		self :	Any ):





           '''simple docstring'''

           return super().test_save_load_optional_components()



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







    """simple docstring"""







    def 				a__			(		self :	Optional[Any] ):





           '''simple docstring'''

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







    def 				a__			(		self :	Tuple ):





           '''simple docstring'''

           __magic_name__      				=    load_image(
               'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/unclip/cat.png' )
           __magic_name__      				=    load_numpy(
               'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
               '/unclip/karlo_v1_alpha_cat_variation_fp16.npy' )

           __magic_name__      				=    UnCLIPImageVariationPipeline.from_pretrained(
               'kakaobrain/karlo-v1-alpha-image-variations'				,    torch_dtype=torch.floataa )
           __magic_name__      				=    pipeline.to(UpperCamelCase_ )
           pipeline.set_progress_bar_config(disable=UpperCamelCase_ )

           __magic_name__      				=    torch.Generator(device='cpu' ).manual_seed(0 )
           __magic_name__      				=    pipeline(
               UpperCamelCase_				,    generator=UpperCamelCase_				,    output_type='np'				,    )

           __magic_name__      				=    output.images[0]

           assert image.shape == (2_5_6, 2_5_6, 3)

           assert_mean_pixel_difference(UpperCamelCase_				,    UpperCamelCase_				,    1_5 )







"""simple docstring"""
import shutil
import tempfile
import unittest
from unittest.mock import patch

from transformers import (
    DefaultFlowCallback,
    IntervalStrategy,
    PrinterCallback,
    ProgressCallback,
    Trainer,
    TrainerCallback,
    TrainingArguments,
    is_torch_available,
)
from transformers.testing_utils import require_torch


if is_torch_available():
 from transformers.trainer import DEFAULT_CALLBACKS

 from .test_trainer import RegressionDataset, RegressionModelConfig, RegressionPreTrainedModel



class        UpperCamelCase__ (				a_):







    """simple docstring"""







    def __init__(		self :	Union[str, Any] ):





           '''simple docstring'''

           __magic_name__      				=    []







    def 				a__			(		self :	Tuple				,    UpperCamelCase_ :	Dict				,    UpperCamelCase_ :	Union[str, Any]				,    UpperCamelCase_ :	int				,    **UpperCamelCase_ :	int ):





           '''simple docstring'''

           self.events.append('on_init_end' )







    def 				a__			(		self :	Any				,    UpperCamelCase_ :	int				,    UpperCamelCase_ :	str				,    UpperCamelCase_ :	List[str]				,    **UpperCamelCase_ :	str ):





           '''simple docstring'''

           self.events.append('on_train_begin' )







    def 				a__			(		self :	str				,    UpperCamelCase_ :	Tuple				,    UpperCamelCase_ :	List[str]				,    UpperCamelCase_ :	Optional[int]				,    **UpperCamelCase_ :	Dict ):





           '''simple docstring'''

           self.events.append('on_train_end' )







    def 				a__			(		self :	List[str]				,    UpperCamelCase_ :	Tuple				,    UpperCamelCase_ :	int				,    UpperCamelCase_ :	Tuple				,    **UpperCamelCase_ :	Optional[Any] ):





           '''simple docstring'''

           self.events.append('on_epoch_begin' )







    def 				a__			(		self :	Optional[int]				,    UpperCamelCase_ :	List[Any]				,    UpperCamelCase_ :	int				,    UpperCamelCase_ :	Tuple				,    **UpperCamelCase_ :	List[Any] ):





           '''simple docstring'''

           self.events.append('on_epoch_end' )







    def 				a__			(		self :	Union[str, Any]				,    UpperCamelCase_ :	Union[str, Any]				,    UpperCamelCase_ :	Any				,    UpperCamelCase_ :	Union[str, Any]				,    **UpperCamelCase_ :	Optional[Any] ):





           '''simple docstring'''

           self.events.append('on_step_begin' )







    def 				a__			(		self :	List[str]				,    UpperCamelCase_ :	Optional[Any]				,    UpperCamelCase_ :	int				,    UpperCamelCase_ :	Any				,    **UpperCamelCase_ :	List[Any] ):





           '''simple docstring'''

           self.events.append('on_step_end' )







    def 				a__			(		self :	List[Any]				,    UpperCamelCase_ :	List[str]				,    UpperCamelCase_ :	int				,    UpperCamelCase_ :	Tuple				,    **UpperCamelCase_ :	Dict ):





           '''simple docstring'''

           self.events.append('on_evaluate' )







    def 				a__			(		self :	str				,    UpperCamelCase_ :	Any				,    UpperCamelCase_ :	Optional[int]				,    UpperCamelCase_ :	Dict				,    **UpperCamelCase_ :	Any ):





           '''simple docstring'''

           self.events.append('on_predict' )







    def 				a__			(		self :	Any				,    UpperCamelCase_ :	Optional[int]				,    UpperCamelCase_ :	Union[str, Any]				,    UpperCamelCase_ :	str				,    **UpperCamelCase_ :	Optional[int] ):





           '''simple docstring'''

           self.events.append('on_save' )







    def 				a__			(		self :	Optional[Any]				,    UpperCamelCase_ :	int				,    UpperCamelCase_ :	str				,    UpperCamelCase_ :	Union[str, Any]				,    **UpperCamelCase_ :	Optional[int] ):





           '''simple docstring'''

           self.events.append('on_log' )







    def 				a__			(		self :	List[Any]				,    UpperCamelCase_ :	Optional[int]				,    UpperCamelCase_ :	List[Any]				,    UpperCamelCase_ :	str				,    **UpperCamelCase_ :	List[Any] ):





           '''simple docstring'''

           self.events.append('on_prediction_step' )



@require_torch
class        UpperCamelCase__ (				unittest.TestCase):







    """simple docstring"""







    def 				a__			(		self :	Optional[Any] ):





           '''simple docstring'''

           __magic_name__      				=    tempfile.mkdtemp()







    def 				a__			(		self :	List[Any] ):





           '''simple docstring'''

           shutil.rmtree(self.output_dir )







    def 				a__			(		self :	List[str]				,    UpperCamelCase_ :	Optional[int]=0				,    UpperCamelCase_ :	List[Any]=0				,    UpperCamelCase_ :	Optional[int]=6_4				,    UpperCamelCase_ :	Optional[int]=6_4				,    UpperCamelCase_ :	List[Any]=None				,    UpperCamelCase_ :	Optional[Any]=False				,    **UpperCamelCase_ :	str ):





           '''simple docstring'''

           __magic_name__      				=    RegressionDataset(length=UpperCamelCase_ )
           __magic_name__      				=    RegressionDataset(length=UpperCamelCase_ )
           __magic_name__      				=    RegressionModelConfig(a=UpperCamelCase_				,    b=UpperCamelCase_ )
           __magic_name__      				=    RegressionPreTrainedModel(UpperCamelCase_ )

           __magic_name__      				=    TrainingArguments(self.output_dir				,    disable_tqdm=UpperCamelCase_				,    report_to=[]				,    **UpperCamelCase_ )
           return Trainer(
               UpperCamelCase_				,    UpperCamelCase_				,    train_dataset=UpperCamelCase_				,    eval_dataset=UpperCamelCase_				,    callbacks=UpperCamelCase_				,    )







    def 				a__			(		self :	Optional[int]				,    UpperCamelCase_ :	str				,    UpperCamelCase_ :	Dict ):





           '''simple docstring'''

           self.assertEqual(len(UpperCamelCase_ )				,    len(UpperCamelCase_ ) )

           # Order doesn't matter
           __magic_name__      				=    sorted(UpperCamelCase_				,    key=lambda UpperCamelCase_ : cb.__name__ if isinstance(UpperCamelCase_				,    UpperCamelCase_ ) else cb.__class__.__name__ )
           __magic_name__      				=    sorted(UpperCamelCase_				,    key=lambda UpperCamelCase_ : cb.__name__ if isinstance(UpperCamelCase_				,    UpperCamelCase_ ) else cb.__class__.__name__ )

           for cba, cba in zip(UpperCamelCase_				,    UpperCamelCase_ ):
                  if isinstance(UpperCamelCase_				,    UpperCamelCase_ ) and isinstance(UpperCamelCase_				,    UpperCamelCase_ ):
                         self.assertEqual(UpperCamelCase_				,    UpperCamelCase_ )
                  elif isinstance(UpperCamelCase_				,    UpperCamelCase_ ) and not isinstance(UpperCamelCase_				,    UpperCamelCase_ ):
                         self.assertEqual(UpperCamelCase_				,    cba.__class__ )
                  elif not isinstance(UpperCamelCase_				,    UpperCamelCase_ ) and isinstance(UpperCamelCase_				,    UpperCamelCase_ ):
                         self.assertEqual(cba.__class__				,    UpperCamelCase_ )
                  else:
                         self.assertEqual(UpperCamelCase_				,    UpperCamelCase_ )







    def 				a__			(		self :	Union[str, Any]				,    UpperCamelCase_ :	Any ):





           '''simple docstring'''

           __magic_name__      				=    ['on_init_end', 'on_train_begin']
           __magic_name__      				=    0
           __magic_name__      				=    len(trainer.get_eval_dataloader() )
           __magic_name__      				=    ['on_prediction_step'] * len(trainer.get_eval_dataloader() ) + ['on_log', 'on_evaluate']
           for _ in range(trainer.state.num_train_epochs ):
                  expected_events.append('on_epoch_begin' )
                  for _ in range(UpperCamelCase_ ):
                         step += 1
                         expected_events += ["on_step_begin", "on_step_end"]
                         if step % trainer.args.logging_steps == 0:
                                expected_events.append('on_log' )
                         if trainer.args.evaluation_strategy == IntervalStrategy.STEPS and step % trainer.args.eval_steps == 0:
                                expected_events += evaluation_events.copy()
                         if step % trainer.args.save_steps == 0:
                                expected_events.append('on_save' )
                  expected_events.append('on_epoch_end' )
                  if trainer.args.evaluation_strategy == IntervalStrategy.EPOCH:
                         expected_events += evaluation_events.copy()
           expected_events += ["on_log", "on_train_end"]
           return expected_events







    def 				a__			(		self :	List[Any] ):





           '''simple docstring'''

           __magic_name__      				=    self.get_trainer()
           __magic_name__      				=    DEFAULT_CALLBACKS.copy() + [ProgressCallback]
           self.check_callbacks_equality(trainer.callback_handler.callbacks				,    UpperCamelCase_ )

           # Callbacks passed at init are added to the default callbacks
           __magic_name__      				=    self.get_trainer(callbacks=[MyTestTrainerCallback] )
           expected_callbacks.append(UpperCamelCase_ )
           self.check_callbacks_equality(trainer.callback_handler.callbacks				,    UpperCamelCase_ )

           # TrainingArguments.disable_tqdm controls if use ProgressCallback or PrinterCallback
           __magic_name__      				=    self.get_trainer(disable_tqdm=UpperCamelCase_ )
           __magic_name__      				=    DEFAULT_CALLBACKS.copy() + [PrinterCallback]
           self.check_callbacks_equality(trainer.callback_handler.callbacks				,    UpperCamelCase_ )







    def 				a__			(		self :	List[str] ):





           '''simple docstring'''

           __magic_name__      				=    DEFAULT_CALLBACKS.copy() + [ProgressCallback]
           __magic_name__      				=    self.get_trainer()

           # We can add, pop, or remove by class name
           trainer.remove_callback(UpperCamelCase_ )
           expected_callbacks.remove(UpperCamelCase_ )
           self.check_callbacks_equality(trainer.callback_handler.callbacks				,    UpperCamelCase_ )

           __magic_name__      				=    self.get_trainer()
           __magic_name__      				=    trainer.pop_callback(UpperCamelCase_ )
           self.assertEqual(cb.__class__				,    UpperCamelCase_ )
           self.check_callbacks_equality(trainer.callback_handler.callbacks				,    UpperCamelCase_ )

           trainer.add_callback(UpperCamelCase_ )
           expected_callbacks.insert(0				,    UpperCamelCase_ )
           self.check_callbacks_equality(trainer.callback_handler.callbacks				,    UpperCamelCase_ )

           # We can also add, pop, or remove by instance
           __magic_name__      				=    self.get_trainer()
           __magic_name__      				=    trainer.callback_handler.callbacks[0]
           trainer.remove_callback(UpperCamelCase_ )
           expected_callbacks.remove(UpperCamelCase_ )
           self.check_callbacks_equality(trainer.callback_handler.callbacks				,    UpperCamelCase_ )

           __magic_name__      				=    self.get_trainer()
           __magic_name__      				=    trainer.callback_handler.callbacks[0]
           __magic_name__      				=    trainer.pop_callback(UpperCamelCase_ )
           self.assertEqual(UpperCamelCase_				,    UpperCamelCase_ )
           self.check_callbacks_equality(trainer.callback_handler.callbacks				,    UpperCamelCase_ )

           trainer.add_callback(UpperCamelCase_ )
           expected_callbacks.insert(0				,    UpperCamelCase_ )
           self.check_callbacks_equality(trainer.callback_handler.callbacks				,    UpperCamelCase_ )







    def 				a__			(		self :	Any ):





           '''simple docstring'''

           import warnings

           # XXX: for now ignore scatter_gather warnings in this test since it's not relevant to what's being tested
           warnings.simplefilter(action='ignore'				,    category=UpperCamelCase_ )

           __magic_name__      				=    self.get_trainer(callbacks=[MyTestTrainerCallback] )
           trainer.train()
           __magic_name__      				=    trainer.callback_handler.callbacks[-2].events
           self.assertEqual(UpperCamelCase_				,    self.get_expected_events(UpperCamelCase_ ) )

           # Independent log/save/eval
           __magic_name__      				=    self.get_trainer(callbacks=[MyTestTrainerCallback]				,    logging_steps=5 )
           trainer.train()
           __magic_name__      				=    trainer.callback_handler.callbacks[-2].events
           self.assertEqual(UpperCamelCase_				,    self.get_expected_events(UpperCamelCase_ ) )

           __magic_name__      				=    self.get_trainer(callbacks=[MyTestTrainerCallback]				,    save_steps=5 )
           trainer.train()
           __magic_name__      				=    trainer.callback_handler.callbacks[-2].events
           self.assertEqual(UpperCamelCase_				,    self.get_expected_events(UpperCamelCase_ ) )

           __magic_name__      				=    self.get_trainer(callbacks=[MyTestTrainerCallback]				,    eval_steps=5				,    evaluation_strategy='steps' )
           trainer.train()
           __magic_name__      				=    trainer.callback_handler.callbacks[-2].events
           self.assertEqual(UpperCamelCase_				,    self.get_expected_events(UpperCamelCase_ ) )

           __magic_name__      				=    self.get_trainer(callbacks=[MyTestTrainerCallback]				,    evaluation_strategy='epoch' )
           trainer.train()
           __magic_name__      				=    trainer.callback_handler.callbacks[-2].events
           self.assertEqual(UpperCamelCase_				,    self.get_expected_events(UpperCamelCase_ ) )

           # A bit of everything
           __magic_name__      				=    self.get_trainer(
               callbacks=[MyTestTrainerCallback]				,    logging_steps=3				,    save_steps=1_0				,    eval_steps=5				,    evaluation_strategy='steps'				,    )
           trainer.train()
           __magic_name__      				=    trainer.callback_handler.callbacks[-2].events
           self.assertEqual(UpperCamelCase_				,    self.get_expected_events(UpperCamelCase_ ) )

           # warning should be emitted for duplicated callbacks
           with patch('transformers.trainer_callback.logger.warning' ) as warn_mock:
                  __magic_name__      				=    self.get_trainer(
                      callbacks=[MyTestTrainerCallback, MyTestTrainerCallback]				,    )
                  assert str(UpperCamelCase_ ) in warn_mock.call_args[0][0]