


'''simple docstring'''

import argparse
import dataclasses
import json
import logging
import os
import shutil
from typing import List, Optional

import datasets
from accelerate import Accelerator
from datasets import load_dataset
from finetuning import finetune
from tqdm.auto import tqdm

import transformers
from transformers import AutoConfig, set_seed
from transformers.trainer_utils import IntervalStrategy


lowerCAmelCase				:str        =      logging.getLogger(__name__)

lowerCAmelCase				:List[Any]        =      '''pytorch_model.bin'''
@dataclasses.dataclass
class _lowerCamelCase	:



		'''simple docstring'''







		A_     :      str				   =    dataclasses.field(
		    metadata={"""help""": """Path to pretrained model or model identifier from huggingface.co/models."""}							)
		A_     :      Optional[str]				   =    dataclasses.field(
		    default=lowercase__		,					metadata={"""help""": """Where do you want to store the pretrained models downloaded from huggingface.co."""}		,					)







@dataclasses.dataclass
class _lowerCamelCase	:



		'''simple docstring'''







		A_     :      str				   =    dataclasses.field(metadata={"""help""": """A csv or a json file containing the training data."""}							)
		A_     :      str				   =    dataclasses.field(metadata={"""help""": """A csv or a json file containing the data to predict on."""}							)
		A_     :      Optional[str]				   =    dataclasses.field(
		    default=lowercase__		,					metadata={"""help""": """A csv or a json file containing the validation data."""}							)
		A_     :      Optional[str]				   =    dataclasses.field(
		    default=lowercase__		,					metadata={"""help""": """The name of the task to train on."""}		,					)
		A_     :      Optional[List[str]]				   =    dataclasses.field(
		    default=lowercase__		,					metadata={"""help""": """The list of labels for the task."""}							)







@dataclasses.dataclass
class _lowerCamelCase	:



		'''simple docstring'''







		A_     :      str				   =    dataclasses.field(
		    metadata={"""help""": """The output directory where the model predictions and checkpoints will be written."""}							)
		A_     :      Optional[str]				   =    dataclasses.field(
		    default="""accuracy"""		,					metadata={"""help""": """The evaluation metric used for the task."""}							)
		A_     :      Optional[str]				   =    dataclasses.field(
		    default="""no"""		,					metadata={
		        """help""": """The evaluation strategy to adopt during training. Possible values are: [\"no\", \"step\", \"epoch]"""
		    }		,					)
		A_     :      Optional[int]				   =    dataclasses.field(
		    default=10		,					metadata={"""help""": """Number of evaluation calls with no improvement after which training will be stopped."""}		,					)
		A_     :      Optional[float]				   =    dataclasses.field(
		    default=0.0		,					metadata={
		        """help""": """How much the specified evaluation metric must improve to satisfy early stopping conditions."""
		    }		,					)
		A_     :      Optional[bool]				   =    dataclasses.field(
		    default=lowercase__		,					metadata={"""help""": """Whether to filter the pseudo-labeled data based on the confidence score."""}		,					)
		A_     :      Optional[bool]				   =    dataclasses.field(
		    default=lowercase__		,					metadata={"""help""": """Whether to filter the pseudo-labeled data based on the validation performance."""}		,					)
		A_     :      Optional[bool]				   =    dataclasses.field(
		    default=lowercase__		,					metadata={"""help""": """Whether to fine-tune on labeled data after pseudo training."""}		,					)
		A_     :      Optional[float]				   =    dataclasses.field(
		    default=0.0		,					metadata={"""help""": """Confidence threshold for pseudo-labeled data filtering."""}		,					)
		A_     :      Optional[int]				   =    dataclasses.field(
		    default=1_00		,					metadata={"""help""": """Number of evaluation calls with no improvement after which training will be stopped."""}		,					)
		A_     :      Optional[int]				   =    dataclasses.field(
		    default=lowercase__		,					metadata={"""help""": """Random seed for initialization."""}		,					)







def lowerCamelCase      (      lowerCAmelCase    :     Tuple		,		lowerCAmelCase    :     Dict		,		lowerCAmelCase    :     Any		,		lowerCAmelCase    :     int		,		lowerCAmelCase    :     List[str]		,		lowerCAmelCase    :     Union[str, Any]		):





	"""simple docstring"""







	__magic_name__	:   List[Any]					     =					datasets.concatenate_datasets([infer_input, infer_output]		,		axis=1		)

	if args.do_filter_by_confidence:
		__magic_name__	:   List[Any]					     =					dataset.filter(lambda lowerCAmelCase		: example["probability"] > args.confidence_threshold		)

	if args.do_filter_by_val_performance:
		assert eval_result >= 0.0 and eval_result <= 1.0
		__magic_name__	:   Any					     =					int(eval_result * len(lowerCAmelCase		)		)
		print(lowerCAmelCase		)
		__magic_name__	:   Any					     =					dataset.sort('probability'		,		reverse=lowerCAmelCase		)
		__magic_name__	:   Dict					     =					dataset.select(range(lowerCAmelCase		)		)

	__magic_name__	:   Dict					     =					dataset.remove_columns(['label', 'probability']		)
	__magic_name__	:   Tuple					     =					dataset.rename_column('prediction'		,		'label'		)
	__magic_name__	:   List[Any]					     =					dataset.map(lambda lowerCAmelCase		: {"label": idalabel[example["label"]]}		)
	__magic_name__	:   Any					     =					dataset.shuffle(seed=args.seed		)

	__magic_name__	:   List[str]					     =					os.path.join(lowerCAmelCase		,		f'train_pseudo.{args.data_file_extension}'		)
	if args.data_file_extension == "csv":
		dataset.to_csv(lowerCAmelCase		,		index=lowerCAmelCase		)
	else:
		dataset.to_json(lowerCAmelCase		)





def lowerCamelCase      (      lowerCAmelCase    :     List[str]		,		lowerCAmelCase    :     List[Any]		,		lowerCAmelCase    :     Any		,		lowerCAmelCase    :     Optional[Any]		,		**lowerCAmelCase    :     Optional[int]		):





	"""simple docstring"""







	__magic_name__	:   int					     =					Accelerator()
	# Make one log on every process with the configuration for debugging.
	logging.basicConfig(
	    format='%(asctime)s - %(levelname)s - %(name)s - %(message)s'		,		datefmt='%m/%d/%Y %H:%M:%S'		,		level=logging.INFO		,		)
	logger.info(accelerator.state		)

	# Setup logging, we only want one process per machine to log things on the
	# screen. accelerator.is_local_main_process is only True for one process per
	# machine.
	logger.setLevel(logging.INFO if accelerator.is_local_main_process else logging.ERROR		)

	if accelerator.is_local_main_process:
		datasets.utils.logging.set_verbosity_warning()
		transformers.utils.logging.set_verbosity_info()
	else:
		datasets.utils.logging.set_verbosity_error()
		transformers.utils.logging.set_verbosity_error()

	__magic_name__	:   List[str]					     =					STModelArguments(model_name_or_path=lowerCAmelCase		)
	__magic_name__	:   Dict					     =					STDataArguments(train_file=lowerCAmelCase		,		infer_file=lowerCAmelCase		)
	__magic_name__	:   int					     =					STTrainingArguments(output_dir=lowerCAmelCase		)
	__magic_name__	:   int					     =					argparse.Namespace()

	for arg_class in (model_args, data_args, training_args):
		for key, value in vars(lowerCAmelCase		).items():
			setattr(lowerCAmelCase		,		lowerCAmelCase		,		lowerCAmelCase		)

	for key, value in kwargs.items():
		if hasattr(lowerCAmelCase		,		lowerCAmelCase		):
			setattr(lowerCAmelCase		,		lowerCAmelCase		,		lowerCAmelCase		)

    # Sanity checks
	__magic_name__	:   int					     =					{}
	__magic_name__	:   str					     =					None

	# You need to provide the training data and the data to predict on
	assert args.train_file is not None
	assert args.infer_file is not None
	__magic_name__	:   int					     =					args.train_file
	__magic_name__	:   Optional[Any]					     =					args.infer_file

	if args.evaluation_strategy != IntervalStrategy.NO.value:
		assert args.eval_file is not None
		__magic_name__	:   Dict					     =					args.eval_file

	for key in data_files:
		__magic_name__	:   List[str]					     =					data_files[key].split('.'		)[-1]
		assert extension in ["csv", "json"], f'`{key}_file` should be a csv or a json file.'
		if args.data_file_extension is None:
			__magic_name__	:   List[Any]					     =					extension
		else:
			assert extension == args.data_file_extension, f'`{key}_file` should be a {args.data_file_extension} file`.'

	assert (
	    args.eval_metric in datasets.list_metrics()
	), f'{args.eval_metric} not in the list of supported metrics {datasets.list_metrics()}.'

	# If passed along, set the training seed now.
	if args.seed is not None:
		set_seed(args.seed		)

	logger.info('Creating the initial data directory for self-training...'		)
	__magic_name__	:   List[Any]					     =					f'{args.output_dir}/self-train_iter-{{}}'.format
	__magic_name__	:   List[Any]					     =					data_dir_format(0		)

	if accelerator.is_main_process:
		if args.output_dir is not None:
			os.makedirs(args.output_dir		,		exist_ok=lowerCAmelCase		)
			os.makedirs(lowerCAmelCase		,		exist_ok=lowerCAmelCase		)
	accelerator.wait_for_everyone()

	__magic_name__	:   Optional[int]					     =					None
	__magic_name__	:   Dict					     =					None
	__magic_name__	:   Any					     =					0
	__magic_name__	:   Tuple					     =					False
	# Show the progress bar
	__magic_name__	:   List[Any]					     =					tqdm(range(args.max_selftrain_iterations		)		,		disable=not accelerator.is_local_main_process		)

	# Self-train
	for iteration in range(0		,		int(args.max_selftrain_iterations		)		):
		__magic_name__	:   List[Any]					     =					data_dir_format(lowerCAmelCase		)
		assert os.path.exists(lowerCAmelCase		)

		# Stage 1: initial fine-tuning for iteration = 0 or pseudo-training for
		# iteration > 0
		__magic_name__	:   int					     =					os.path.join(lowerCAmelCase		,		'stage-1'		)
		__magic_name__	:   Optional[Any]					     =					{
		    'accelerator': accelerator,
		    'model_name_or_path': args.model_name_or_path,
		    'cache_dir': args.cache_dir,
		    'do_train': True,
		    'train_file': data_files['train'] if iteration == 0 else data_files['train_pseudo'],
		    'do_eval': True if args.eval_file is not None else False,
		    'eval_file': data_files['eval'],
		    'do_predict': True,
		    'infer_file': data_files['infer'],
		    'task_name': args.task_name,
		    'label_list': args.label_list,
		    'output_dir': current_output_dir,
		    'eval_metric': args.eval_metric,
		    'evaluation_strategy': args.evaluation_strategy,
		    'early_stopping_patience': args.early_stopping_patience,
		    'early_stopping_threshold': args.early_stopping_threshold,
		    'seed': args.seed,
		}
		# Add additional training arguments
		for key, value in kwargs.items():
			if key not in arguments_dict and not hasattr(lowerCAmelCase		,		lowerCAmelCase		):
				arguments_dict.update({key: value}		)

		__magic_name__	:   Optional[int]					     =					os.path.join(lowerCAmelCase		,		'best-checkpoint'		,		lowerCAmelCase		)
		if os.path.exists(lowerCAmelCase		):
			logger.info(
			    'Found existing model checkpoint at %s. Skipping self-training: iteration: %d, stage: 1.'		,		lowerCAmelCase		,		lowerCAmelCase		,		)
		else:
			logger.info('***** Running self-training: iteration: %d, stage: 1 *****'		,		lowerCAmelCase		)
			finetune(**lowerCAmelCase		)
			accelerator.wait_for_everyone()
			assert os.path.exists(lowerCAmelCase		)
			logger.info('Self-training job completed: iteration: %d, stage: 1.'		,		lowerCAmelCase		)

		if iteration > 0 and args.finetune_on_labeled_data:
			# Stage 2 (optional): fine-tuning on the original labeled data
			__magic_name__	:   Any					     =					os.path.join(lowerCAmelCase		,		'best-checkpoint'		)
			__magic_name__	:   Optional[int]					     =					os.path.join(lowerCAmelCase		,		'stage-2'		)
			# Update arguments_dict
			__magic_name__	:   Tuple					     =					model_path
			__magic_name__	:   Union[str, Any]					     =					data_files['train']
			__magic_name__	:   Optional[Any]					     =					current_output_dir

			__magic_name__	:   int					     =					os.path.join(lowerCAmelCase		,		'best-checkpoint'		,		lowerCAmelCase		)
			if os.path.exists(lowerCAmelCase		):
				logger.info(
				    'Found existing model checkpoint at %s. Skipping self-training: iteration: %d, stage: 2.'		,		lowerCAmelCase		,		lowerCAmelCase		,		)
			else:
				logger.info('***** Running self-training: iteration: %d, stage: 2 *****'		,		lowerCAmelCase		)
				finetune(**lowerCAmelCase		)
				accelerator.wait_for_everyone()
				assert os.path.exists(lowerCAmelCase		)
				logger.info('Self-training job completed: iteration: %d, stage: 2.'		,		lowerCAmelCase		)

		__magic_name__	:   Tuple					     =					iteration
		__magic_name__	:   List[str]					     =					data_dir_format(iteration + 1		)

		__magic_name__	:   Any					     =					AutoConfig.from_pretrained(os.path.join(lowerCAmelCase		,		'best-checkpoint'		)		)
		__magic_name__	:   List[Any]					     =					config.idalabel
		__magic_name__	:   int					     =					os.path.join(lowerCAmelCase		,		'eval_results_best-checkpoint.json'		)
		__magic_name__	:   Union[str, Any]					     =					os.path.join(lowerCAmelCase		,		'test_results_best-checkpoint.json'		)
		assert os.path.exists(lowerCAmelCase		)

		with open(lowerCAmelCase		,		'r'		) as f:
			__magic_name__	:   str					     =					float(json.load(lowerCAmelCase		)[args.eval_metric]		)
		__magic_name__	:   Optional[Any]					     =					os.path.join(lowerCAmelCase		,		'infer_output_best-checkpoint.csv'		)
		assert os.path.exists(lowerCAmelCase		)
		# Loading the dataset from local csv or json files.
		__magic_name__	:   Tuple					     =					load_dataset(args.data_file_extension		,		data_files={'data': data_files['infer']}		)['data']
		__magic_name__	:   Optional[int]					     =					load_dataset('csv'		,		data_files={'data': infer_output_file}		)['data']

		if accelerator.is_main_process:
			os.makedirs(lowerCAmelCase		,		exist_ok=lowerCAmelCase		)
			shutil.copy(lowerCAmelCase		,		os.path.join(lowerCAmelCase		,		f'eval_results_iter-{iteration}.json'		)		)
			if os.path.exists(lowerCAmelCase		):
				shutil.copy(lowerCAmelCase		,		os.path.join(lowerCAmelCase		,		f'test_results_iter-{iteration}.json'		)		)
			create_pseudo_labeled_data(lowerCAmelCase		,		lowerCAmelCase		,		lowerCAmelCase		,		lowerCAmelCase		,		lowerCAmelCase		,		lowerCAmelCase		)
		accelerator.wait_for_everyone()

		__magic_name__	:   Union[str, Any]					     =					os.path.join(lowerCAmelCase		,		f'train_pseudo.{args.data_file_extension}'		)

		if args.evaluation_strategy != IntervalStrategy.NO.value:
			__magic_name__	:   Optional[Any]					     =					eval_result

			if best_iteration is None:
				__magic_name__	:   Dict					     =					new_iteration
				__magic_name__	:   Tuple					     =					new_eval_result
			else:
				if new_eval_result - best_eval_result > args.early_stopping_threshold:
					__magic_name__	:   List[str]					     =					new_iteration
					__magic_name__	:   Dict					     =					new_eval_result
					__magic_name__	:   Any					     =					0
				else:
					if new_eval_result == best_eval_result:
						__magic_name__	:   Union[str, Any]					     =					new_iteration
						__magic_name__	:   Union[str, Any]					     =					new_eval_result
					early_stopping_patience_counter += 1

				if early_stopping_patience_counter >= args.early_stopping_patience:
					__magic_name__	:   List[Any]					     =					True

		progress_bar.update(1		)

		if should_training_stop:
			break

	if best_iteration is not None:
		# Save the best iteration
		logger.info('Best iteration: %d'		,		lowerCAmelCase		)
		logger.info('Best evaluation result: %s = %f'		,		args.eval_metric		,		lowerCAmelCase		)
		accelerator.wait_for_everyone()
		if accelerator.is_main_process:
			shutil.copy(
			    os.path.join(lowerCAmelCase		,		f'eval_results_iter-{iteration}.json'		)		,		os.path.join(lowerCAmelCase		,		'eval_results_best-iteration.json'		)		,		)
	else:
		# Assume that the last iteration is the best
		logger.info('Best iteration: %d'		,		args.max_selftrain_iterations - 1		)
		logger.info('Best evaluation result: %s = %f'		,		args.eval_metric		,		lowerCAmelCase		)
		accelerator.wait_for_everyone()
		if accelerator.is_main_process:
			shutil.copy(
			    os.path.join(lowerCAmelCase		,		f'eval_results_iter-{args.max_selftrain_iterations - 1}.json'		)		,		os.path.join(lowerCAmelCase		,		'eval_results_best-iteration.json'		)		,		)



'''simple docstring'''

import gc
import random
import unittest

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

from diffusers import (
    AutoencoderKL,
    DiffusionPipeline,
    EulerDiscreteScheduler,
    StableDiffusionXLImgaImgPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import floats_tensor, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import (
    IMAGE_TO_IMAGE_IMAGE_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()
class _lowerCamelCase	(						lowercase__		,					lowercase__		,					unittest.TestCase							):



		'''simple docstring'''







		A_     :      int				   =    StableDiffusionXLImgaImgPipeline
		A_     :      str				   =    TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"""height""", """width"""}
		A_     :      Optional[Any]				   =    PipelineTesterMixin.required_optional_params - {"""latents"""}
		A_     :      Any				   =    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
		A_     :      Any				   =    IMAGE_TO_IMAGE_IMAGE_PARAMS
		A_     :      Optional[int]				   =    IMAGE_TO_IMAGE_IMAGE_PARAMS







		def        __lowerCAmelCase		( self		:						Dict      )  ->			Optional[int]:
			torch.manual_seed(0      )
			__magic_name__	:   int					     =					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')  ,					attention_head_dim=(2, 4)  ,					use_linear_projection=_A  ,					addition_embed_type='text_time'  ,					addition_time_embed_dim=8  ,					transformer_layers_per_block=(1, 2)  ,					projection_class_embeddings_input_dim=80  ,					cross_attention_dim=64  ,					)
			__magic_name__	:   str					     =					EulerDiscreteScheduler(
			    beta_start=0.0_0085  ,					beta_end=0.012  ,					steps_offset=1  ,					beta_schedule='scaled_linear'  ,					timestep_spacing='leading'  ,					)
			torch.manual_seed(0      )
			__magic_name__	:   str					     =					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  ,					sample_size=128  ,					)
			torch.manual_seed(0      )
			__magic_name__	:   Any					     =					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  ,					hidden_act='gelu'  ,					projection_dim=32  ,					)
			__magic_name__	:   Dict					     =					CLIPTextModel(_A      )
			__magic_name__	:   List[str]					     =					CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'  ,					local_files_only=_A      )

			__magic_name__	:   Optional[Any]					     =					CLIPTextModelWithProjection(_A      )
			__magic_name__	:   Dict					     =					CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'  ,					local_files_only=_A      )

			__magic_name__	:   List[Any]					     =					{
			    'unet': unet,
			    'scheduler': scheduler,
			    'vae': vae,
			    'text_encoder': text_encoder,
			    'tokenizer': tokenizer,
			    'text_encoder_2': text_encoder_a,
			    'tokenizer_2': tokenizer_a,
			    # "safety_checker": None,
			    # "feature_extractor": None,
			}
			return components







		def        __lowerCAmelCase		( self		:						List[Any]  ,					_A		:						List[str]  ,					_A		:						Any=0      )  ->			Union[str, Any]:
			__magic_name__	:   Any					     =					floats_tensor((1, 3, 32, 32)  ,					rng=random.Random(_A      )      ).to(_A      )
			__magic_name__	:   Dict					     =					image / 2 + 0.5
			if str(_A      ).startswith('mps'      ):
				__magic_name__	:   Any					     =					torch.manual_seed(_A      )
			else:
				__magic_name__	:   int					     =					torch.Generator(device=_A      ).manual_seed(_A      )
			__magic_name__	:   List[Any]					     =					{
			    'prompt': 'A painting of a squirrel eating a burger',
			    'image': image,
			    'generator': generator,
			    'num_inference_steps': 2,
			    'guidance_scale': 5.0,
			    'output_type': 'numpy',
			    'strength': 0.75,
			}
			return inputs







		def        __lowerCAmelCase		( self		:						Optional[Any]      )  ->			List[Any]:
			__magic_name__	:   str					     =					'cpu'  # ensure determinism for the device-dependent torch.Generator
			__magic_name__	:   str					     =					self.get_dummy_components()
			__magic_name__	:   Any					     =					StableDiffusionXLImgaImgPipeline(**_A      )
			__magic_name__	:   List[Any]					     =					sd_pipe.to(_A      )
			sd_pipe.set_progress_bar_config(disable=_A      )

			__magic_name__	:   Any					     =					self.get_dummy_inputs(_A      )
			__magic_name__	:   Optional[int]					     =					sd_pipe(**_A      ).images
			__magic_name__	:   str					     =					image[0, -3:, -3:, -1]

			assert image.shape == (1, 32, 32, 3)

			__magic_name__	:   Any					     =					np.array([0.4656, 0.4840, 0.4439, 0.6698, 0.5574, 0.4524, 0.5799, 0.5943, 0.5165]      )

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







		def        __lowerCAmelCase		( self		:						Optional[int]      )  ->			Optional[Any]:
			super().test_attention_slicing_forward_pass(expected_max_diff=3E-3      )







		def        __lowerCAmelCase		( self		:						List[Any]      )  ->			int:
			super().test_inference_batch_single_identical(expected_max_diff=3E-3      )







		def        __lowerCAmelCase		( self		:						Any      )  ->			Union[str, Any]:
			pass







		def        __lowerCAmelCase		( self		:						List[Any]      )  ->			Optional[int]:
			__magic_name__	:   Dict					     =					self.get_dummy_components()
			__magic_name__	:   Optional[Any]					     =					StableDiffusionXLImgaImgPipeline(**_A      )
			__magic_name__	:   List[Any]					     =					sd_pipe.to(_A      )
			__magic_name__	:   str					     =					sd_pipe.to(_A      )
			sd_pipe.set_progress_bar_config(disable=_A      )

			# forward without prompt embeds
			__magic_name__	:   Union[str, Any]					     =					self.get_dummy_inputs(_A      )
			__magic_name__	:   Union[str, Any]					     =					3 * ['this is a negative prompt']
			__magic_name__	:   List[str]					     =					negative_prompt
			__magic_name__	:   int					     =					3 * [inputs['prompt']]

			__magic_name__	:   Tuple					     =					sd_pipe(**_A      )
			__magic_name__	:   str					     =					output.images[0, -3:, -3:, -1]

			# forward with prompt embeds
			__magic_name__	:   Optional[Any]					     =					self.get_dummy_inputs(_A      )
			__magic_name__	:   Tuple					     =					3 * ['this is a negative prompt']
			__magic_name__	:   List[str]					     =					3 * [inputs.pop('prompt'      )]

			(
			    (
			    __magic_name__
			)     ,    (
			    __magic_name__
			)     ,    (
			    __magic_name__
			)     ,    (
			    __magic_name__
			)     ,    
			)	:   List[Any]					     =					sd_pipe.encode_prompt(_A  ,					negative_prompt=_A      )

			__magic_name__	:   Tuple					     =					sd_pipe(
			    **_A  ,					prompt_embeds=_A  ,					negative_prompt_embeds=_A  ,					pooled_prompt_embeds=_A  ,					negative_pooled_prompt_embeds=_A  ,					)
			__magic_name__	:   int					     =					output.images[0, -3:, -3:, -1]

			# make sure that it's equal
			assert np.abs(image_slice_a.flatten() - image_slice_a.flatten()      ).max() < 1E-4







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



		'''simple docstring'''







		def        __lowerCAmelCase		( self		:						List[Any]      )  ->			Tuple:
			super().tearDown()
			gc.collect()
			torch.cuda.empty_cache()







		def        __lowerCAmelCase		( self		:						str  ,					_A		:						Optional[int]  ,					_A		:						Optional[Any]="cpu"  ,					_A		:						List[str]=torch.floataa  ,					_A		:						Any=0      )  ->			str:
			__magic_name__	:   List[str]					     =					torch.Generator(device=_A      ).manual_seed(_A      )
			__magic_name__	:   Optional[Any]					     =					np.random.RandomState(_A      ).standard_normal((1, 4, 64, 64)      )
			__magic_name__	:   Union[str, Any]					     =					torch.from_numpy(_A      ).to(device=_A  ,					dtype=_A      )
			__magic_name__	:   Optional[int]					     =					{
			    'prompt': 'a photograph of an astronaut riding a horse',
			    'latents': latents,
			    'generator': generator,
			    'num_inference_steps': 3,
			    'guidance_scale': 7.5,
			    'output_type': 'numpy',
			}
			return inputs







		def        __lowerCAmelCase		( self		:						Union[str, Any]      )  ->			Dict:
			__magic_name__	:   str					     =					DiffusionPipeline.from_pretrained('stabilityai/stable-diffusion-2-base'      )
			pipe.to(_A      )
			pipe.set_progress_bar_config(disable=_A      )

			__magic_name__	:   Optional[int]					     =					self.get_inputs(_A      )
			__magic_name__	:   Union[str, Any]					     =					pipe(**_A      ).images
			__magic_name__	:   Any					     =					image[0, -3:, -3:, -1].flatten()

			assert image.shape == (1, 512, 512, 3)
			__magic_name__	:   List[Any]					     =					np.array([0.4_9493, 0.4_7896, 0.4_0798, 0.5_4214, 0.5_3212, 0.4_8202, 0.4_7656, 0.4_6329, 0.4_8506]      )
			assert np.abs(image_slice - expected_slice      ).max() < 7E-3