"""simple docstring"""





import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import XLMRobertaTokenizerFast

from diffusers import DDIMScheduler, KandinskyImgaImgPipeline, KandinskyPriorPipeline, UNetaDConditionModel, VQModel
from diffusers.pipelines.kandinsky.text_encoder import MCLIPConfig, MultilingualCLIP
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()


class   UpperCAmelCase_			(					_a				,							unittest.TestCase):
      lowerCamelCase__	:					str										=						KandinskyImgaImgPipeline
      lowerCamelCase__	:					List[str]										=						["prompt", "image_embeds", "negative_image_embeds", "image"]
      lowerCamelCase__	:					Optional[int]										=						[
          "prompt",
          "negative_prompt",
          "image_embeds",
          "negative_image_embeds",
          "image",
      ]
      lowerCamelCase__	:					Union[str, Any]										=						[
          "generator",
          "height",
          "width",
          "strength",
          "guidance_scale",
          "negative_prompt",
          "num_inference_steps",
          "return_dict",
          "guidance_scale",
          "num_images_per_prompt",
          "output_type",
          "return_dict",
      ]
      lowerCamelCase__	:					Optional[int]										=						False


      @property
      def 	_UpperCAmelCase (     self							) ->      List[str]:
             return 3_2


      @property
      def 	_UpperCAmelCase (     self							) ->      List[Any]:
             return 3_2


      @property
      def 	_UpperCAmelCase (     self							) ->      Optional[int]:
             return self.time_input_dim


      @property
      def 	_UpperCAmelCase (     self							) ->      Union[str, Any]:
             return self.time_input_dim * 4


      @property
      def 	_UpperCAmelCase (     self							) ->      int:
             return 1_0_0


      @property
      def 	_UpperCAmelCase (     self							) ->      Optional[Any]:
             lowercase__    :	int     					=			XLMRobertaTokenizerFast.from_pretrained('YiYiXu/tiny-random-mclip-base'							)
             return tokenizer


      @property
      def 	_UpperCAmelCase (     self							) ->      Optional[int]:
             torch.manual_seed(0							)
             lowercase__    :	Optional[Any]     					=			MCLIPConfig(
                 numDims=self.cross_attention_dim	,    transformerDimensions=self.text_embedder_hidden_size	,    hidden_size=self.text_embedder_hidden_size	,    intermediate_size=3_7	,    num_attention_heads=4	,    num_hidden_layers=5	,    vocab_size=1_0_0_5	,    )

             lowercase__    :	Tuple     					=			MultilingualCLIP(a							)
             lowercase__    :	List[str]     					=			text_encoder.eval()

             return text_encoder


      @property
      def 	_UpperCAmelCase (     self							) ->      Optional[Any]:
             torch.manual_seed(0							)

             lowercase__    :	List[str]     					=			{
                 'in_channels': 4,
                 # Out channels is double in channels because predicts mean and variance
                 'out_channels': 8,
                 'addition_embed_type': 'text_image',
                 '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,
                 'encoder_hid_dim': self.text_embedder_hidden_size,
                 'encoder_hid_dim_type': 'text_image_proj',
                 'cross_attention_dim': self.cross_attention_dim,
                 'attention_head_dim': 4,
                 'resnet_time_scale_shift': 'scale_shift',
                 'class_embed_type': None,
             }

             lowercase__    :	Any     					=			UNetaDConditionModel(**a							)
             return model


      @property
      def 	_UpperCAmelCase (     self							) ->      Optional[Any]:
             return {
                 "block_out_channels": [3_2, 6_4],
                 "down_block_types": ["DownEncoderBlock2D", "AttnDownEncoderBlock2D"],
                 "in_channels": 3,
                 "latent_channels": 4,
                 "layers_per_block": 1,
                 "norm_num_groups": 8,
                 "norm_type": "spatial",
                 "num_vq_embeddings": 1_2,
                 "out_channels": 3,
                 "up_block_types": [
                     "AttnUpDecoderBlock2D",
                     "UpDecoderBlock2D",
                 ],
                 "vq_embed_dim": 4,
             }


      @property
      def 	_UpperCAmelCase (     self							) ->      Union[str, Any]:
             torch.manual_seed(0							)
             lowercase__    :	Any     					=			VQModel(**self.dummy_movq_kwargs							)
             return model


      def 	_UpperCAmelCase (     self							) ->      Optional[Any]:
             lowercase__    :	List[Any]     					=			self.dummy_text_encoder
             lowercase__    :	int     					=			self.dummy_tokenizer
             lowercase__    :	Tuple     					=			self.dummy_unet
             lowercase__    :	Tuple     					=			self.dummy_movq

             lowercase__    :	Union[str, Any]     					=			{
                 'num_train_timesteps': 1_0_0_0,
                 'beta_schedule': 'linear',
                 'beta_start': 0.00_085,
                 'beta_end': 0.012,
                 'clip_sample': False,
                 'set_alpha_to_one': False,
                 'steps_offset': 0,
                 'prediction_type': 'epsilon',
                 'thresholding': False,
             }

             lowercase__    :	Dict     					=			DDIMScheduler(**a							)

             lowercase__    :	int     					=			{
                 'text_encoder': text_encoder,
                 'tokenizer': tokenizer,
                 'unet': unet,
                 'scheduler': scheduler,
                 'movq': movq,
             }

             return components


      def 	_UpperCAmelCase (     self	,    a	,    a=0							) ->      Tuple:
             lowercase__    :	Any     					=			floats_tensor((1, self.cross_attention_dim)	,    rng=random.Random(a							)							).to(a							)
             lowercase__    :	List[str]     					=			floats_tensor((1, self.cross_attention_dim)	,    rng=random.Random(seed + 1							)							).to(a							)
             # create init_image
             lowercase__    :	str     					=			floats_tensor((1, 3, 6_4, 6_4)	,    rng=random.Random(a							)							).to(a							)
             lowercase__    :	Optional[int]     					=			image.cpu().permute(0	,    2	,    3	,    1							)[0]
             lowercase__    :	List[Any]     					=			Image.fromarray(np.uinta(a							)							).convert('RGB'							).resize((2_5_6, 2_5_6)							)

             if str(a							).startswith('mps'							):
                    lowercase__    :	Tuple     					=			torch.manual_seed(a							)
             else:
                    lowercase__    :	Tuple     					=			torch.Generator(device=a							).manual_seed(a							)
             lowercase__    :	str     					=			{
                 'prompt': 'horse',
                 'image': init_image,
                 'image_embeds': image_embeds,
                 'negative_image_embeds': negative_image_embeds,
                 'generator': generator,
                 'height': 6_4,
                 'width': 6_4,
                 'num_inference_steps': 1_0,
                 'guidance_scale': 7.0,
                 'strength': 0.2,
                 'output_type': 'np',
             }
             return inputs


      def 	_UpperCAmelCase (     self							) ->      List[str]:
             lowercase__    :	Any     					=			'cpu'

             lowercase__    :	List[str]     					=			self.get_dummy_components()

             lowercase__    :	List[str]     					=			self.pipeline_class(**a							)
             lowercase__    :	Dict     					=			pipe.to(a							)

             pipe.set_progress_bar_config(disable=a							)

             lowercase__    :	Optional[int]     					=			pipe(**self.get_dummy_inputs(a							)							)
             lowercase__    :	Tuple     					=			output.images

             lowercase__    :	Union[str, Any]     					=			pipe(
                 **self.get_dummy_inputs(a							)	,    return_dict=a	,    )[0]

             lowercase__    :	Dict     					=			image[0, -3:, -3:, -1]
             lowercase__    :	Optional[Any]     					=			image_from_tuple[0, -3:, -3:, -1]

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

             lowercase__    :	Tuple     					=			np.array(
                 [0.61_474_943, 0.6_073_539, 0.43_308_544, 0.5_928_269, 0.47_493_595, 0.46_755_973, 0.4_613_838, 0.45_368_797, 0.50_119_233]							)
             assert (
                 np.abs(image_slice.flatten() - expected_slice							).max() < 1e-2
             ), f""" expected_slice {expected_slice}, but got {image_slice.flatten()}"""
             assert (
                 np.abs(image_from_tuple_slice.flatten() - expected_slice							).max() < 1e-2
             ), f""" expected_slice {expected_slice}, but got {image_from_tuple_slice.flatten()}"""





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


      def 	_UpperCAmelCase (     self							) ->      int:
             # clean up the VRAM after each test
             super().tearDown()
             gc.collect()
             torch.cuda.empty_cache()


      def 	_UpperCAmelCase (     self							) ->      int:
             lowercase__    :	Optional[int]     					=			load_numpy(
                 'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
                 '/kandinsky/kandinsky_img2img_frog.npy'							)

             lowercase__    :	Tuple     					=			load_image(
                 'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main' '/kandinsky/cat.png'							)
             lowercase__    :	str     					=			'A red cartoon frog, 4k'

             lowercase__    :	Dict     					=			KandinskyPriorPipeline.from_pretrained(
                 'kandinsky-community/kandinsky-2-1-prior'	,    torch_dtype=torch.floataa							)
             pipe_prior.to(a							)

             lowercase__    :	int     					=			KandinskyImgaImgPipeline.from_pretrained(
                 'kandinsky-community/kandinsky-2-1'	,    torch_dtype=torch.floataa							)
             lowercase__    :	Union[str, Any]     					=			pipeline.to(a							)

             pipeline.set_progress_bar_config(disable=a							)

             lowercase__    :	str     					=			torch.Generator(device='cpu'							).manual_seed(0							)
             lowercase__					,   lowercase__    :	str     					=			pipe_prior(
                 a	,    generator=a	,    num_inference_steps=5	,    negative_prompt=''	,    ).to_tuple()

             lowercase__    :	Optional[Any]     					=			pipeline(
                 a	,    image=a	,    image_embeds=a	,    negative_image_embeds=a	,    generator=a	,    num_inference_steps=1_0_0	,    height=7_6_8	,    width=7_6_8	,    strength=0.2	,    output_type='np'	,    )

             lowercase__    :	Union[str, Any]     					=			output.images[0]

             assert image.shape == (7_6_8, 7_6_8, 3)

             assert_mean_pixel_difference(a	,    a							)


"""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):


      def __init__(     self							) ->      Any:
             lowercase__    :	Tuple     					=			[]


      def 	_UpperCAmelCase (     self	,    a	,    a	,    a	,    **a							) ->      Any:
             self.events.append('on_init_end'							)


      def 	_UpperCAmelCase (     self	,    a	,    a	,    a	,    **a							) ->      Optional[int]:
             self.events.append('on_train_begin'							)


      def 	_UpperCAmelCase (     self	,    a	,    a	,    a	,    **a							) ->      List[str]:
             self.events.append('on_train_end'							)


      def 	_UpperCAmelCase (     self	,    a	,    a	,    a	,    **a							) ->      int:
             self.events.append('on_epoch_begin'							)


      def 	_UpperCAmelCase (     self	,    a	,    a	,    a	,    **a							) ->      Optional[Any]:
             self.events.append('on_epoch_end'							)


      def 	_UpperCAmelCase (     self	,    a	,    a	,    a	,    **a							) ->      int:
             self.events.append('on_step_begin'							)


      def 	_UpperCAmelCase (     self	,    a	,    a	,    a	,    **a							) ->      str:
             self.events.append('on_step_end'							)


      def 	_UpperCAmelCase (     self	,    a	,    a	,    a	,    **a							) ->      int:
             self.events.append('on_evaluate'							)


      def 	_UpperCAmelCase (     self	,    a	,    a	,    a	,    **a							) ->      Tuple:
             self.events.append('on_predict'							)


      def 	_UpperCAmelCase (     self	,    a	,    a	,    a	,    **a							) ->      Union[str, Any]:
             self.events.append('on_save'							)


      def 	_UpperCAmelCase (     self	,    a	,    a	,    a	,    **a							) ->      List[str]:
             self.events.append('on_log'							)


      def 	_UpperCAmelCase (     self	,    a	,    a	,    a	,    **a							) ->      Any:
             self.events.append('on_prediction_step'							)





@require_torch
class   UpperCAmelCase_			(					unittest.TestCase):


      def 	_UpperCAmelCase (     self							) ->      str:
             lowercase__    :	str     					=			tempfile.mkdtemp()


      def 	_UpperCAmelCase (     self							) ->      Dict:
             shutil.rmtree(self.output_dir							)


      def 	_UpperCAmelCase (     self	,    a=0	,    a=0	,    a=6_4	,    a=6_4	,    a=None	,    a=False	,    **a							) ->      int:
             # disable_tqdm in TrainingArguments has a flaky default since it depends on the level of logging. We make sure
             # its set to False since the tests later on depend on its value.
             lowercase__    :	str     					=			RegressionDataset(length=a							)
             lowercase__    :	Any     					=			RegressionDataset(length=a							)
             lowercase__    :	Optional[Any]     					=			RegressionModelConfig(a=a	,    b=a							)
             lowercase__    :	Union[str, Any]     					=			RegressionPreTrainedModel(a							)

             lowercase__    :	Tuple     					=			TrainingArguments(self.output_dir	,    disable_tqdm=a	,    report_to=[]	,    **a							)
             return Trainer(
                 a	,    a	,    train_dataset=a	,    eval_dataset=a	,    callbacks=a	,    )


      def 	_UpperCAmelCase (     self	,    a	,    a							) ->      Union[str, Any]:
             self.assertEqual(len(a							)	,    len(a							)							)

             # Order doesn't matter
             lowercase__    :	Optional[int]     					=			sorted(a	,    key=lambda a							: cb.__name__ if isinstance(a	,    a							) else cb.__class__.__name__							)
             lowercase__    :	Tuple     					=			sorted(a	,    key=lambda a							: cb.__name__ if isinstance(a	,    a							) else cb.__class__.__name__							)

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


      def 	_UpperCAmelCase (     self	,    a							) ->      Optional[Any]:
             lowercase__    :	Dict     					=			['on_init_end', 'on_train_begin']
             lowercase__    :	List[Any]     					=			0
             lowercase__    :	Optional[int]     					=			len(trainer.get_eval_dataloader()							)
             lowercase__    :	Tuple     					=			['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(a							):
                           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 	_UpperCAmelCase (     self							) ->      Union[str, Any]:
             lowercase__    :	int     					=			self.get_trainer()
             lowercase__    :	str     					=			DEFAULT_CALLBACKS.copy() + [ProgressCallback]
             self.check_callbacks_equality(trainer.callback_handler.callbacks	,    a							)

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

             # TrainingArguments.disable_tqdm controls if use ProgressCallback or PrinterCallback
             lowercase__    :	List[Any]     					=			self.get_trainer(disable_tqdm=a							)
             lowercase__    :	Optional[Any]     					=			DEFAULT_CALLBACKS.copy() + [PrinterCallback]
             self.check_callbacks_equality(trainer.callback_handler.callbacks	,    a							)


      def 	_UpperCAmelCase (     self							) ->      Any:
             lowercase__    :	int     					=			DEFAULT_CALLBACKS.copy() + [ProgressCallback]
             lowercase__    :	List[str]     					=			self.get_trainer()

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

             lowercase__    :	Optional[Any]     					=			self.get_trainer()
             lowercase__    :	List[Any]     					=			trainer.pop_callback(a							)
             self.assertEqual(cb.__class__	,    a							)
             self.check_callbacks_equality(trainer.callback_handler.callbacks	,    a							)

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

             # We can also add, pop, or remove by instance
             lowercase__    :	int     					=			self.get_trainer()
             lowercase__    :	List[str]     					=			trainer.callback_handler.callbacks[0]
             trainer.remove_callback(a							)
             expected_callbacks.remove(a							)
             self.check_callbacks_equality(trainer.callback_handler.callbacks	,    a							)

             lowercase__    :	Tuple     					=			self.get_trainer()
             lowercase__    :	Dict     					=			trainer.callback_handler.callbacks[0]
             lowercase__    :	Union[str, Any]     					=			trainer.pop_callback(a							)
             self.assertEqual(a	,    a							)
             self.check_callbacks_equality(trainer.callback_handler.callbacks	,    a							)

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


      def 	_UpperCAmelCase (     self							) ->      Tuple:
             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=a							)

             lowercase__    :	Optional[Any]     					=			self.get_trainer(callbacks=[MyTestTrainerCallback]							)
             trainer.train()
             lowercase__    :	Any     					=			trainer.callback_handler.callbacks[-2].events
             self.assertEqual(a	,    self.get_expected_events(a							)							)

             # Independent log/save/eval
             lowercase__    :	List[str]     					=			self.get_trainer(callbacks=[MyTestTrainerCallback]	,    logging_steps=5							)
             trainer.train()
             lowercase__    :	int     					=			trainer.callback_handler.callbacks[-2].events
             self.assertEqual(a	,    self.get_expected_events(a							)							)

             lowercase__    :	Union[str, Any]     					=			self.get_trainer(callbacks=[MyTestTrainerCallback]	,    save_steps=5							)
             trainer.train()
             lowercase__    :	Union[str, Any]     					=			trainer.callback_handler.callbacks[-2].events
             self.assertEqual(a	,    self.get_expected_events(a							)							)

             lowercase__    :	List[str]     					=			self.get_trainer(callbacks=[MyTestTrainerCallback]	,    eval_steps=5	,    evaluation_strategy='steps'							)
             trainer.train()
             lowercase__    :	Optional[int]     					=			trainer.callback_handler.callbacks[-2].events
             self.assertEqual(a	,    self.get_expected_events(a							)							)

             lowercase__    :	int     					=			self.get_trainer(callbacks=[MyTestTrainerCallback]	,    evaluation_strategy='epoch'							)
             trainer.train()
             lowercase__    :	str     					=			trainer.callback_handler.callbacks[-2].events
             self.assertEqual(a	,    self.get_expected_events(a							)							)

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

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

