



"""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        __snake_case       (	_lowercase  ,     unittest.TestCase):
  snake_case__				:   List[str]  			=						KandinskyImgaImgPipeline
  snake_case__				:   Any  			=						["prompt", "image_embeds", "negative_image_embeds", "image"]
  snake_case__				:   str  			=						[
      "prompt",
      "negative_prompt",
      "image_embeds",
      "negative_image_embeds",
      "image",
  ]
  snake_case__				:   Optional[int]  			=						[
      "generator",
      "height",
      "width",
      "strength",
      "guidance_scale",
      "negative_prompt",
      "num_inference_steps",
      "return_dict",
      "guidance_scale",
      "num_images_per_prompt",
      "output_type",
      "return_dict",
  ]
  snake_case__				:   List[Any]  			=						False




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






         """simple docstring"""






         return 3_2




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






         """simple docstring"""






         return 3_2




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






         """simple docstring"""






         return self.time_input_dim




  @property
  def 						SCREAMING_SNAKE_CASE				( self     :						Optional[Any]		):






         """simple docstring"""






         return self.time_input_dim * 4




  @property
  def 						SCREAMING_SNAKE_CASE				( self     :						List[str]		):






         """simple docstring"""






         return 1_0_0




  @property
  def 						SCREAMING_SNAKE_CASE				( self     :						str		):






         """simple docstring"""






         _lowerCamelCase						:				List[Any]    			= XLMRobertaTokenizerFast.from_pretrained('''YiYiXu/tiny-random-mclip-base'''		)
         return tokenizer




  @property
  def 						SCREAMING_SNAKE_CASE				( self     :						str		):






         """simple docstring"""






         torch.manual_seed(0		)
         _lowerCamelCase						:				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						,					)

         _lowerCamelCase						:				Tuple    			= MultilingualCLIP(__lowerCAmelCase		)
         _lowerCamelCase						:				Optional[int]    			= text_encoder.eval()

         return text_encoder




  @property
  def 						SCREAMING_SNAKE_CASE				( self     :						str		):






         """simple docstring"""






         torch.manual_seed(0		)

         _lowerCamelCase						:				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,
         }

         _lowerCamelCase						:				List[Any]    			= UNetaDConditionModel(**__lowerCAmelCase		)
         return model




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






         """simple docstring"""






         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 						SCREAMING_SNAKE_CASE				( self     :						List[str]		):






         """simple docstring"""






         torch.manual_seed(0		)
         _lowerCamelCase						:				Optional[Any]    			= VQModel(**self.dummy_movq_kwargs		)
         return model




  def 						SCREAMING_SNAKE_CASE				( self     :						Optional[int]		):






         """simple docstring"""






         _lowerCamelCase						:				Dict    			= self.dummy_text_encoder
         _lowerCamelCase						:				Tuple    			= self.dummy_tokenizer
         _lowerCamelCase						:				Dict    			= self.dummy_unet
         _lowerCamelCase						:				int    			= self.dummy_movq

         _lowerCamelCase						:				Any    			= {
             '''num_train_timesteps''': 1_0_0_0,
             '''beta_schedule''': '''linear''',
             '''beta_start''': 0.0_00_85,
             '''beta_end''': 0.0_12,
             '''clip_sample''': False,
             '''set_alpha_to_one''': False,
             '''steps_offset''': 0,
             '''prediction_type''': '''epsilon''',
             '''thresholding''': False,
         }

         _lowerCamelCase						:				Optional[Any]    			= DDIMScheduler(**__lowerCAmelCase		)

         _lowerCamelCase						:				List[str]    			= {
             '''text_encoder''': text_encoder,
             '''tokenizer''': tokenizer,
             '''unet''': unet,
             '''scheduler''': scheduler,
             '''movq''': movq,
         }

         return components




  def 						SCREAMING_SNAKE_CASE				( self     :						str						,					__lowerCAmelCase     :						Any						,					__lowerCAmelCase     :						int=0		):






         """simple docstring"""






         _lowerCamelCase						:				int    			= floats_tensor((1, self.cross_attention_dim)						,					rng=random.Random(__lowerCAmelCase		)		).to(__lowerCAmelCase		)
         _lowerCamelCase						:				Optional[Any]    			= floats_tensor((1, self.cross_attention_dim)						,					rng=random.Random(seed + 1		)		).to(__lowerCAmelCase		)
         # create init_image
         _lowerCamelCase						:				Any    			= floats_tensor((1, 3, 6_4, 6_4)						,					rng=random.Random(__lowerCAmelCase		)		).to(__lowerCAmelCase		)
         _lowerCamelCase						:				Optional[Any]    			= image.cpu().permute(0						,					2						,					3						,					1		)[0]
         _lowerCamelCase						:				Dict    			= Image.fromarray(np.uinta(__lowerCAmelCase		)		).convert('''RGB'''		).resize((2_5_6, 2_5_6)		)

         if str(__lowerCAmelCase		).startswith('''mps'''		):
                _lowerCamelCase						:				Dict    			= torch.manual_seed(__lowerCAmelCase		)
         else:
                _lowerCamelCase						:				int    			= torch.Generator(device=__lowerCAmelCase		).manual_seed(__lowerCAmelCase		)
         _lowerCamelCase						:				Tuple    			= {
             '''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 						SCREAMING_SNAKE_CASE				( self     :						str		):






         """simple docstring"""






         _lowerCamelCase						:				Optional[int]    			= '''cpu'''

         _lowerCamelCase						:				List[Any]    			= self.get_dummy_components()

         _lowerCamelCase						:				Optional[int]    			= self.pipeline_class(**__lowerCAmelCase		)
         _lowerCamelCase						:				Optional[Any]    			= pipe.to(__lowerCAmelCase		)

         pipe.set_progress_bar_config(disable=__lowerCAmelCase		)

         _lowerCamelCase						:				Dict    			= pipe(**self.get_dummy_inputs(__lowerCAmelCase		)		)
         _lowerCamelCase						:				Optional[int]    			= output.images

         _lowerCamelCase						:				int    			= pipe(
             **self.get_dummy_inputs(__lowerCAmelCase		)						,					return_dict=__lowerCAmelCase						,					)[0]

         _lowerCamelCase						:				List[Any]    			= image[0, -3:, -3:, -1]
         _lowerCamelCase						:				List[str]    			= image_from_tuple[0, -3:, -3:, -1]

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

         _lowerCamelCase						:				List[Any]    			= np.array(
             [0.61_47_49_43, 0.6_07_35_39, 0.43_30_85_44, 0.5_92_82_69, 0.47_49_35_95, 0.46_75_59_73, 0.4_61_38_38, 0.45_36_87_97, 0.50_11_92_33]		)
         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        __snake_case       (	unittest.TestCase):




  def 						SCREAMING_SNAKE_CASE				( self     :						List[str]		):






         """simple docstring"""






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




  def 						SCREAMING_SNAKE_CASE				( self     :						Tuple		):






         """simple docstring"""






         _lowerCamelCase						:				Any    			= load_numpy(
             '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
             '''/kandinsky/kandinsky_img2img_frog.npy'''		)

         _lowerCamelCase						:				Dict    			= load_image(
             '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main''' '''/kandinsky/cat.png'''		)
         _lowerCamelCase						:				List[Any]    			= '''A red cartoon frog, 4k'''

         _lowerCamelCase						:				int    			= KandinskyPriorPipeline.from_pretrained(
             '''kandinsky-community/kandinsky-2-1-prior'''						,					torch_dtype=torch.floataa		)
         pipe_prior.to(__lowerCAmelCase		)

         _lowerCamelCase						:				List[Any]    			= KandinskyImgaImgPipeline.from_pretrained(
             '''kandinsky-community/kandinsky-2-1'''						,					torch_dtype=torch.floataa		)
         _lowerCamelCase						:				Optional[Any]    			= pipeline.to(__lowerCAmelCase		)

         pipeline.set_progress_bar_config(disable=__lowerCAmelCase		)

         _lowerCamelCase						:				Dict    			= torch.Generator(device='''cpu'''		).manual_seed(0		)
         _lowerCamelCase			,    _lowerCamelCase						:				List[str]    			= pipe_prior(
             __lowerCAmelCase						,					generator=__lowerCAmelCase						,					num_inference_steps=5						,					negative_prompt=''''''						,					).to_tuple()

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

         _lowerCamelCase						:				List[str]    			= output.images[0]

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

         assert_mean_pixel_difference(__lowerCAmelCase						,					__lowerCAmelCase		)



import json
import os
import subprocess
import unittest
from ast import literal_eval

import pytest
from parameterized import parameterized_class

from . import is_sagemaker_available


if is_sagemaker_available():
      from sagemaker import Session, TrainingJobAnalytics
      from sagemaker.huggingface import HuggingFace





@pytest.mark.skipif(
    literal_eval(os.getenv('TEST_SAGEMAKER'		,				'False'     )     ) is not True		,				reason='Skipping test because should only be run when releasing minor transformers version'		,				)
@pytest.mark.usefixtures('sm_env'     )
@parameterized_class(
    [
        {
            'framework': 'pytorch',
            'script': 'run_glue.py',
            'model_name_or_path': 'distilbert-base-cased',
            'instance_type': 'ml.g4dn.xlarge',
            'results': {'train_runtime': 6_50, 'eval_accuracy': 0.6, 'eval_loss': 0.9},
        },
        {
            'framework': 'tensorflow',
            'script': 'run_tf.py',
            'model_name_or_path': 'distilbert-base-cased',
            'instance_type': 'ml.g4dn.xlarge',
            'results': {'train_runtime': 6_00, 'eval_accuracy': 0.3, 'eval_loss': 0.9},
        },
    ]     )
class      A							(				unittest.TestCase     ):


   def 			lowercase_      (self     :						int		)						->				Optional[Any]:
         """simple docstring"""



         if self.framework == "pytorch":
               subprocess.run(
                   f"""cp ./examples/pytorch/text-classification/run_glue.py {self.env.test_path}/run_glue.py""".split()       ,  encoding="utf-8"       ,  check=__UpperCAmelCase       ,  )
         assert hasattr(self       ,  "env"		)


   def 			lowercase_      (self     :						List[Any]       ,  __UpperCAmelCase     :						Optional[int]=1		)						->				Dict:
         """simple docstring"""



         return HuggingFace(
             entry_point=self.script       ,  source_dir=self.env.test_path       ,  role=self.env.role       ,  image_uri=self.env.image_uri       ,  base_job_name=f"""{self.env.base_job_name}-single"""       ,  instance_count=__UpperCAmelCase       ,  instance_type=self.instance_type       ,  debugger_hook_config=__UpperCAmelCase       ,  hyperparameters={**self.env.hyperparameters, "model_name_or_path": self.model_name_or_path}       ,  metric_definitions=self.env.metric_definitions       ,  py_version="py36"       ,  )


   def 			lowercase_      (self     :						Optional[Any]       ,  __UpperCAmelCase     :						Tuple		)						->				Optional[int]:
         """simple docstring"""



         TrainingJobAnalytics(__UpperCAmelCase		).export_csv(f"""{self.env.test_path}/{job_name}_metrics.csv"""		)





   def 			lowercase_      (self     :						Any		)						->				Optional[Any]:
         """simple docstring"""



         UpperCAmelCase__								=  self.create_estimator()

         # run training
         estimator.fit()

         # result dataframe
         UpperCAmelCase__								=  TrainingJobAnalytics(estimator.latest_training_job.name		).dataframe()

         # extract kpis
         UpperCAmelCase__								=  list(result_metrics_df[result_metrics_df.metric_name == "eval_accuracy"]["value"]		)
         UpperCAmelCase__								=  list(result_metrics_df[result_metrics_df.metric_name == "eval_loss"]["value"]		)
         # get train time from SageMaker job, this includes starting, preprocessing, stopping
         UpperCAmelCase__								=  (
             Session().describe_training_job(estimator.latest_training_job.name		).get("TrainingTimeInSeconds"       ,  9_9_9_9_9_9		)
         )

         # assert kpis
         assert train_runtime <= self.results["train_runtime"]
         assert all(t >= self.results["eval_accuracy"] for t in eval_accuracy		)
         assert all(t <= self.results["eval_loss"] for t in eval_loss		)

         # dump tests result into json file to share in PR
         with open(f"""{estimator.latest_training_job.name}.json"""       ,  "w"		) as outfile:
               json.dump({"train_time": train_runtime, "eval_accuracy": eval_accuracy, "eval_loss": eval_loss}       ,  __UpperCAmelCase		)



