
import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    StableDiffusionSAGPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()





class   __UpperCamelCase      (				UpperCamelCase_      ,		UpperCamelCase_      ,		unittest.TestCase    ):






     """simple docstring"""







     lowerCAmelCase_									=       StableDiffusionSAGPipeline
     lowerCAmelCase_									=       TEXT_TO_IMAGE_PARAMS
     lowerCAmelCase_									=       TEXT_TO_IMAGE_BATCH_PARAMS
     lowerCAmelCase_									=       TEXT_TO_IMAGE_IMAGE_PARAMS
     lowerCAmelCase_									=       TEXT_TO_IMAGE_IMAGE_PARAMS
     lowerCAmelCase_									=       False



     def       UpperCAmelCase__				(							self				:				str      ):







       """simple docstring"""




       torch.manual_seed(0      )
       __SCREAMING_SNAKE_CASE	:     Optional[Any] =						UNetaDConditionModel(
           block_out_channels=(32, 64)			,		layers_per_block=2			,		sample_size=32			,		in_channels=4			,		out_channels=4			,		down_block_types=('''DownBlock2D''', '''CrossAttnDownBlock2D''')			,		up_block_types=('''CrossAttnUpBlock2D''', '''UpBlock2D''')			,		cross_attention_dim=32			,		)
       __SCREAMING_SNAKE_CASE	:     Optional[int] =						DDIMScheduler(
           beta_start=0.0_00_85			,		beta_end=0.0_12			,		beta_schedule='''scaled_linear'''			,		clip_sample=A_			,		set_alpha_to_one=A_			,		)
       torch.manual_seed(0      )
       __SCREAMING_SNAKE_CASE	:     int =						AutoencoderKL(
           block_out_channels=[32, 64]			,		in_channels=3			,		out_channels=3			,		down_block_types=['''DownEncoderBlock2D''', '''DownEncoderBlock2D''']			,		up_block_types=['''UpDecoderBlock2D''', '''UpDecoderBlock2D''']			,		latent_channels=4			,		)
       torch.manual_seed(0      )
       __SCREAMING_SNAKE_CASE	:     int =						CLIPTextConfig(
           bos_token_id=0			,		eos_token_id=2			,		hidden_size=32			,		intermediate_size=37			,		layer_norm_eps=1e-05			,		num_attention_heads=4			,		num_hidden_layers=5			,		pad_token_id=1			,		vocab_size=1000			,		)
       __SCREAMING_SNAKE_CASE	:     List[Any] =						CLIPTextModel(A_      )
       __SCREAMING_SNAKE_CASE	:     int =						CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''      )

       __SCREAMING_SNAKE_CASE	:     Optional[Any] =						{
           '''unet''': unet,
           '''scheduler''': scheduler,
           '''vae''': vae,
           '''text_encoder''': text_encoder,
           '''tokenizer''': tokenizer,
           '''safety_checker''': None,
           '''feature_extractor''': None,
       }
       return components



     def       UpperCAmelCase__				(							self				:				List[Any]			,		_A				:				List[str]			,		_A				:				Dict=0      ):







       """simple docstring"""




       if str(A_      ).startswith('''mps'''      ):
         __SCREAMING_SNAKE_CASE	:     str =						torch.manual_seed(A_      )
       else:
         __SCREAMING_SNAKE_CASE	:     str =						torch.Generator(device=A_      ).manual_seed(A_      )
       __SCREAMING_SNAKE_CASE	:     Optional[int] =						{
           '''prompt''': '''.''',
           '''generator''': generator,
           '''num_inference_steps''': 2,
           '''guidance_scale''': 1.0,
           '''sag_scale''': 1.0,
           '''output_type''': '''numpy''',
       }
       return inputs




     def       UpperCAmelCase__				(							self				:				int      ):







       """simple docstring"""




       super().test_inference_batch_single_identical(expected_max_diff=3e-3      )






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






     """simple docstring"""







     def       UpperCAmelCase__				(							self				:				Dict      ):







       """simple docstring"""




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



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







       """simple docstring"""




       __SCREAMING_SNAKE_CASE	:     Optional[int] =						StableDiffusionSAGPipeline.from_pretrained('''CompVis/stable-diffusion-v1-4'''      )
       __SCREAMING_SNAKE_CASE	:     Optional[Any] =						sag_pipe.to(A_      )
       sag_pipe.set_progress_bar_config(disable=A_      )

       __SCREAMING_SNAKE_CASE	:     Optional[Any] =						'''.'''
       __SCREAMING_SNAKE_CASE	:     Union[str, Any] =						torch.manual_seed(0      )
       __SCREAMING_SNAKE_CASE	:     List[str] =						sag_pipe(
           [prompt]			,		generator=A_			,		guidance_scale=7.5			,		sag_scale=1.0			,		num_inference_steps=20			,		output_type='''np'''      )

       __SCREAMING_SNAKE_CASE	:     List[Any] =						output.images

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

       assert image.shape == (1, 512, 512, 3)
       __SCREAMING_SNAKE_CASE	:     Any =						np.array([0.15_68, 0.17_38, 0.16_95, 0.16_93, 0.15_07, 0.17_05, 0.15_47, 0.17_51, 0.19_49]      )

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



     def       UpperCAmelCase__				(							self				:				List[str]      ):







       """simple docstring"""




       __SCREAMING_SNAKE_CASE	:     Optional[int] =						StableDiffusionSAGPipeline.from_pretrained('''stabilityai/stable-diffusion-2-1-base'''      )
       __SCREAMING_SNAKE_CASE	:     Any =						sag_pipe.to(A_      )
       sag_pipe.set_progress_bar_config(disable=A_      )

       __SCREAMING_SNAKE_CASE	:     Optional[Any] =						'''.'''
       __SCREAMING_SNAKE_CASE	:     str =						torch.manual_seed(0      )
       __SCREAMING_SNAKE_CASE	:     Tuple =						sag_pipe(
           [prompt]			,		generator=A_			,		guidance_scale=7.5			,		sag_scale=1.0			,		num_inference_steps=20			,		output_type='''np'''      )

       __SCREAMING_SNAKE_CASE	:     Any =						output.images

       __SCREAMING_SNAKE_CASE	:     Optional[int] =						image[0, -3:, -3:, -1]

       assert image.shape == (1, 512, 512, 3)
       __SCREAMING_SNAKE_CASE	:     int =						np.array([0.34_59, 0.28_76, 0.25_37, 0.30_02, 0.26_71, 0.21_60, 0.30_26, 0.22_62, 0.23_71]      )

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




     def       UpperCAmelCase__				(							self				:				List[Any]      ):







       """simple docstring"""




       __SCREAMING_SNAKE_CASE	:     Optional[Any] =						StableDiffusionSAGPipeline.from_pretrained('''stabilityai/stable-diffusion-2-1-base'''      )
       __SCREAMING_SNAKE_CASE	:     int =						sag_pipe.to(A_      )
       sag_pipe.set_progress_bar_config(disable=A_      )

       __SCREAMING_SNAKE_CASE	:     List[str] =						'''.'''
       __SCREAMING_SNAKE_CASE	:     str =						torch.manual_seed(0      )
       __SCREAMING_SNAKE_CASE	:     Dict =						sag_pipe(
           [prompt]			,		width=768			,		height=512			,		generator=A_			,		guidance_scale=7.5			,		sag_scale=1.0			,		num_inference_steps=20			,		output_type='''np'''			,		)

       __SCREAMING_SNAKE_CASE	:     Union[str, Any] =						output.images

       assert image.shape == (1, 512, 768, 3)





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''': 650, '''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''': 600, '''eval_accuracy''': 0.3, '''eval_loss''': 0.9},
        },
    ]						)
class 				__snake_case						(     unittest.TestCase						):




							def    UpperCAmelCase__			(       self     :       int):
									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=A_				,				)
									assert hasattr(self				,				'''env''')




							def    UpperCAmelCase__			(       self     :       Union[str, Any]				,				A_     :       str=1):
									# creates estimator
									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=A_				,				instance_type=self.instance_type				,				debugger_hook_config=A_				,				hyperparameters={**self.env.hyperparameters, '''model_name_or_path''': self.model_name_or_path}				,				metric_definitions=self.env.metric_definitions				,				py_version='''py36'''				,				)




							def    UpperCAmelCase__			(       self     :       List[str]				,				A_     :       Optional[Any]):
									TrainingJobAnalytics(A_).export_csv(F"""{self.env.test_path}/{job_name}_metrics.csv""")







							def    UpperCAmelCase__			(       self     :       int):
									# create estimator
									lowerCAmelCase_			:     List[Any]									= self.create_estimator()

									# run training
									estimator.fit()

									# result dataframe
									lowerCAmelCase_			:     Union[str, Any]									= TrainingJobAnalytics(estimator.latest_training_job.name).dataframe()

									# extract kpis
									lowerCAmelCase_			:     Optional[Any]									= list(result_metrics_df[result_metrics_df.metric_name == '''eval_accuracy''']['''value'''])
									lowerCAmelCase_			:     Optional[int]									= list(result_metrics_df[result_metrics_df.metric_name == '''eval_loss''']['''value'''])
									# get train time from SageMaker job, this includes starting, preprocessing, stopping
									lowerCAmelCase_			:     Dict									= (
									    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}				,				A_)





