






'''simple docstring'''




import argparse
import shutil
import time
from json import JSONDecodeError
from logging import getLogger
from pathlib import Path
from typing import Dict, List

import torch
from torch.utils.data import DataLoader
from tqdm import tqdm

from transformers import AutoModelForSeqaSeqLM, AutoTokenizer
from utils import (
    SeqaSeqDataset,
    calculate_bleu,
    calculate_rouge,
    chunks,
    lmap,
    load_json,
    parse_numeric_n_bool_cl_kwargs,
    save_json,
    use_task_specific_params,
    write_txt_file,
)


lowerCAmelCase__          =						getLogger(__name__)







def       __UpperCAmelCase						(  lowerCamelCase_     ,      lowerCamelCase_     ,      lowerCamelCase_     ,      lowerCamelCase_ = 8     ,      lowerCamelCase_ = 1_024     ,      lowerCamelCase_="val"     ,      lowerCamelCase_=None     ,      lowerCamelCase_=False     ,      lowerCamelCase_="summarization"     ,      lowerCamelCase_=None     ,      lowerCamelCase_=1     ,      lowerCamelCase_ = None     ,      lowerCamelCase_=""     ,      **lowerCamelCase_     ,      )							->     Union[str, Any]:
  UpperCamelCase__				:					List[Any]					=					str(SCREAMING_SNAKE_CASE_)
  assert local_rank is not None
  torch.distributed.init_process_group(backend='nccl'     ,      rank=SCREAMING_SNAKE_CASE_)

  UpperCamelCase__				:					str					=					Path(SCREAMING_SNAKE_CASE_)
  UpperCamelCase__				:					Any					=					save_dir.joinpath(f'rank_{local_rank}_output.json')
  torch.cuda.set_device(SCREAMING_SNAKE_CASE_)
  UpperCamelCase__				:					Union[str, Any]					=					AutoModelForSeqaSeqLM.from_pretrained(SCREAMING_SNAKE_CASE_).cuda()
  if fpaa:
    UpperCamelCase__				:					List[Any]					=					model.half()
  # determine if we need to increase num_beams
  use_task_specific_params(SCREAMING_SNAKE_CASE_     ,      SCREAMING_SNAKE_CASE_)  # update config with task specific params
  UpperCamelCase__				:					int					=					generate_kwargs.pop('num_beams'     ,      model.config.num_beams)  # AttributeError risk?
  if num_return_sequences > num_beams:
    UpperCamelCase__				:					int					=					num_return_sequences

  UpperCamelCase__				:					List[Any]					=					AutoTokenizer.from_pretrained(SCREAMING_SNAKE_CASE_)
  logger.info(f'Inferred tokenizer type: {tokenizer.__class__}')  # if this is wrong, check config.model_type.

  if max_source_length is None:
    UpperCamelCase__				:					int					=					tokenizer.model_max_length
  if prefix is None:
    UpperCamelCase__				:					List[Any]					=					prefix or getattr(model.config     ,      'prefix'     ,      '') or ''
  UpperCamelCase__				:					Optional[int]					=					SeqaSeqDataset(
      SCREAMING_SNAKE_CASE_     ,      SCREAMING_SNAKE_CASE_     ,      SCREAMING_SNAKE_CASE_     ,      max_target_length=1_024     ,      type_path=SCREAMING_SNAKE_CASE_     ,      n_obs=SCREAMING_SNAKE_CASE_     ,      prefix=SCREAMING_SNAKE_CASE_     ,      **SCREAMING_SNAKE_CASE_     ,      )
  # I set shuffle=True for a more accurate progress bar.
  # If all the longest samples are first, the prog bar estimate is too high at the beginning.
  UpperCamelCase__				:					List[str]					=					ds.make_sortish_sampler(SCREAMING_SNAKE_CASE_     ,      distributed=SCREAMING_SNAKE_CASE_     ,      add_extra_examples=SCREAMING_SNAKE_CASE_     ,      shuffle=SCREAMING_SNAKE_CASE_)
  UpperCamelCase__				:					List[str]					=					DataLoader(SCREAMING_SNAKE_CASE_     ,      sampler=SCREAMING_SNAKE_CASE_     ,      batch_size=SCREAMING_SNAKE_CASE_     ,      collate_fn=ds.collate_fn)
  UpperCamelCase__				:					Optional[Any]					=					[]
  for batch in tqdm(SCREAMING_SNAKE_CASE_):
    UpperCamelCase__				:					List[str]					=					model.generate(
        input_ids=batch['input_ids'].to(model.device)     ,      attention_mask=batch['attention_mask'].to(model.device)     ,      num_return_sequences=SCREAMING_SNAKE_CASE_     ,      num_beams=SCREAMING_SNAKE_CASE_     ,      **SCREAMING_SNAKE_CASE_     ,      )
    UpperCamelCase__				:					Tuple					=					tokenizer.batch_decode(SCREAMING_SNAKE_CASE_     ,      skip_special_tokens=SCREAMING_SNAKE_CASE_     ,      clean_up_tokenization_spaces=SCREAMING_SNAKE_CASE_)
    UpperCamelCase__				:					List[Any]					=					batch['ids']
    if num_return_sequences > 1:
      UpperCamelCase__				:					str					=					chunks(SCREAMING_SNAKE_CASE_     ,      SCREAMING_SNAKE_CASE_)  # batch size chunks, each of size num_return_seq
    for i, pred in enumerate(SCREAMING_SNAKE_CASE_):
      results.append({'pred': pred, 'id': ids[i].item()})
  save_json(SCREAMING_SNAKE_CASE_     ,      SCREAMING_SNAKE_CASE_)
  return results, sampler.num_replicas







def       __UpperCAmelCase						(  )							->     Optional[Any]:
  UpperCamelCase__				:					Tuple					=					argparse.ArgumentParser(
      epilog='Unspecified args like --num_beams=2 --decoder_start_token_id=4 are passed to model.generate')
  parser.add_argument('--data_dir'     ,      type=SCREAMING_SNAKE_CASE_     ,      help='like cnn_dm/test.source')
  parser.add_argument(
      '--model_name'     ,      type=SCREAMING_SNAKE_CASE_     ,      help='like facebook/bart-large-cnn,t5-base, etc.'     ,      default='sshleifer/distilbart-xsum-12-3'     ,      )
  parser.add_argument('--save_dir'     ,      type=SCREAMING_SNAKE_CASE_     ,      help='where to save'     ,      default='tmp_gen')
  parser.add_argument('--max_source_length'     ,      type=SCREAMING_SNAKE_CASE_     ,      default=SCREAMING_SNAKE_CASE_)
  parser.add_argument(
      '--type_path'     ,      type=SCREAMING_SNAKE_CASE_     ,      default='test'     ,      help='which subset to evaluate typically train/val/test')
  parser.add_argument('--task'     ,      type=SCREAMING_SNAKE_CASE_     ,      default='summarization'     ,      help='used for task_specific_params + metrics')
  parser.add_argument('--bs'     ,      type=SCREAMING_SNAKE_CASE_     ,      default=8     ,      required=SCREAMING_SNAKE_CASE_     ,      help='batch size')
  parser.add_argument(
      '--local_rank'     ,      type=SCREAMING_SNAKE_CASE_     ,      default=-1     ,      required=SCREAMING_SNAKE_CASE_     ,      help='should be passed by distributed.launch')

  parser.add_argument(
      '--n_obs'     ,      type=SCREAMING_SNAKE_CASE_     ,      default=SCREAMING_SNAKE_CASE_     ,      required=SCREAMING_SNAKE_CASE_     ,      help='How many observations. Defaults to all.')
  parser.add_argument(
      '--num_return_sequences'     ,      type=SCREAMING_SNAKE_CASE_     ,      default=1     ,      required=SCREAMING_SNAKE_CASE_     ,      help='How many sequences to return')
  parser.add_argument(
      '--sync_timeout'     ,      type=SCREAMING_SNAKE_CASE_     ,      default=600     ,      required=SCREAMING_SNAKE_CASE_     ,      help='How long should master process wait for other processes to finish.'     ,      )
  parser.add_argument('--src_lang'     ,      type=SCREAMING_SNAKE_CASE_     ,      default=SCREAMING_SNAKE_CASE_     ,      required=SCREAMING_SNAKE_CASE_)
  parser.add_argument('--tgt_lang'     ,      type=SCREAMING_SNAKE_CASE_     ,      default=SCREAMING_SNAKE_CASE_     ,      required=SCREAMING_SNAKE_CASE_)
  parser.add_argument(
      '--prefix'     ,      type=SCREAMING_SNAKE_CASE_     ,      required=SCREAMING_SNAKE_CASE_     ,      default=SCREAMING_SNAKE_CASE_     ,      help='will be added to the begininng of src examples')
  parser.add_argument('--fp16'     ,      action='store_true')
  parser.add_argument('--debug'     ,      action='store_true')
  UpperCamelCase__				:					Optional[int]					=					time.time()
  UpperCamelCase__,	UpperCamelCase__				:					Optional[int]					=					parser.parse_known_args()
  UpperCamelCase__				:					Union[str, Any]					=					parse_numeric_n_bool_cl_kwargs(SCREAMING_SNAKE_CASE_)
  if generate_kwargs and args.local_rank <= 0:
    print(f'parsed the following generate kwargs: {generate_kwargs}')
  UpperCamelCase__				:					Tuple					=					Path(args.save_dir + '_tmp')
  Path(SCREAMING_SNAKE_CASE_).mkdir(exist_ok=SCREAMING_SNAKE_CASE_)  # this handles locking.
  UpperCamelCase__				:					Optional[Any]					=					list(json_save_dir.glob('rank_*.json'))
  if intermediate_files:
    raise ValueError(f'Found files at {json_save_dir} please move or remove them.')
    # In theory, a node could finish and save before another node hits this. If this happens, we can address later.
  UpperCamelCase__				:					Union[str, Any]					=					{}
  if args.src_lang is not None:
    UpperCamelCase__				:					Optional[int]					=					args.src_lang
  if args.tgt_lang is not None:
    UpperCamelCase__				:					Union[str, Any]					=					args.tgt_lang

  Path(args.save_dir).mkdir(exist_ok=SCREAMING_SNAKE_CASE_)
  UpperCamelCase__,	UpperCamelCase__				:					Optional[int]					=					eval_data_dir(
      args.data_dir     ,      SCREAMING_SNAKE_CASE_     ,      args.model_name     ,      type_path=args.type_path     ,      bs=args.bs     ,      fpaa=args.fpaa     ,      task=args.task     ,      local_rank=args.local_rank     ,      n_obs=args.n_obs     ,      max_source_length=args.max_source_length     ,      num_return_sequences=args.num_return_sequences     ,      prefix=args.prefix     ,      dataset_kwargs=SCREAMING_SNAKE_CASE_     ,      **SCREAMING_SNAKE_CASE_     ,      )

  if args.local_rank <= 0:
    UpperCamelCase__				:					List[str]					=					Path(args.save_dir)
    save_dir.mkdir(exist_ok=SCREAMING_SNAKE_CASE_)
    UpperCamelCase__				:					str					=					gather_results_from_each_node(SCREAMING_SNAKE_CASE_     ,      SCREAMING_SNAKE_CASE_     ,      args.sync_timeout)
    UpperCamelCase__				:					int					=					combine_partial_results(SCREAMING_SNAKE_CASE_)
    if args.num_return_sequences > 1:
      UpperCamelCase__				:					Any					=					save_dir.joinpath('pseudolabel_results.json')
      print(f'Saving aggregated results at {save_path}, intermediate in {json_save_dir}/')
      save_json(SCREAMING_SNAKE_CASE_     ,      SCREAMING_SNAKE_CASE_)
      return
    UpperCamelCase__				:					Optional[int]					=					Path(args.data_dir).joinpath(args.type_path + '.target')
    with open(SCREAMING_SNAKE_CASE_) as f:
      UpperCamelCase__				:					Any					=					[x.rstrip() for x in f.readlines()][: len(SCREAMING_SNAKE_CASE_)]

    # Calculate metrics, save metrics,  and save _generations.txt
    UpperCamelCase__				:					Tuple					=					'translation' in args.task
    UpperCamelCase__				:					Optional[int]					=					calculate_bleu if calc_bleu else calculate_rouge
    UpperCamelCase__				:					List[Any]					=					'bleu' if calc_bleu else 'rouge'
    UpperCamelCase__				:					Optional[int]					=					score_fn(SCREAMING_SNAKE_CASE_     ,      SCREAMING_SNAKE_CASE_)
    UpperCamelCase__				:					List[Any]					=					len(SCREAMING_SNAKE_CASE_)
    UpperCamelCase__				:					Optional[Any]					=					time.time() - start_time
    UpperCamelCase__				:					List[str]					=					round(runtime / metrics['n_obs']     ,      4)
    UpperCamelCase__				:					Optional[Any]					=					num_replicas
    # TODO(@stas00): add whatever metadata to metrics
    UpperCamelCase__				:					Any					=					save_dir.joinpath(f'{args.type_path}_{metric_name}.json')
    save_json(SCREAMING_SNAKE_CASE_     ,      SCREAMING_SNAKE_CASE_     ,      indent=SCREAMING_SNAKE_CASE_)
    print(SCREAMING_SNAKE_CASE_)
    write_txt_file(SCREAMING_SNAKE_CASE_     ,      save_dir.joinpath(f'{args.type_path}_generations.txt'))
    if args.debug:
      write_txt_file(SCREAMING_SNAKE_CASE_     ,      save_dir.joinpath(f'{args.type_path}.target'))
    else:
      shutil.rmtree(SCREAMING_SNAKE_CASE_)







def       __UpperCAmelCase						(  lowerCamelCase_)							->     Union[str, Any]:
  UpperCamelCase__				:					Tuple					=					[]
  for partial_result in partial_results:
    records.extend(SCREAMING_SNAKE_CASE_)
  UpperCamelCase__				:					Tuple					=					sorted(SCREAMING_SNAKE_CASE_     ,      key=lambda lowerCamelCase_: x["id"])
  UpperCamelCase__				:					Optional[int]					=					[x['pred'] for x in records]
  return preds







def       __UpperCAmelCase						(  lowerCamelCase_     ,      lowerCamelCase_     ,      lowerCamelCase_)							->     Optional[Any]:
  # WAIT FOR lots of .json files
  UpperCamelCase__				:					Optional[int]					=					time.time()
  logger.info('waiting for all nodes to finish')
  UpperCamelCase__				:					Tuple					=					None
  while (time.time() - start_wait) < timeout:
    UpperCamelCase__				:					Tuple					=					list(save_dir.glob('rank_*.json'))
    if len(SCREAMING_SNAKE_CASE_) < num_replicas:
      continue
    try:
      # make sure all json files are fully saved
      UpperCamelCase__				:					str					=					lmap(SCREAMING_SNAKE_CASE_     ,      SCREAMING_SNAKE_CASE_)
      return json_data
    except JSONDecodeError:
      continue
  else:
    raise TimeoutError('Rank 0 gave up on waiting for other processes')
  # Unreachable


if __name__ == "__main__":
  # Usage for MT:
  run_generate()












'''simple docstring'''




import gc
import random
import tempfile
import unittest

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

from diffusers import AutoencoderKL, DDIMScheduler, LMSDiscreteScheduler, PNDMScheduler, UNetaDConditionModel
from diffusers.pipelines.stable_diffusion_safe import StableDiffusionPipelineSafe as StableDiffusionPipeline
from diffusers.utils import floats_tensor, nightly, torch_device
from diffusers.utils.testing_utils import require_torch_gpu







class       __lowercase					(unittest.TestCase    ):

	def      __UpperCamelCase							(      self   :						List[str]):
			# clean up the VRAM after each test
			super().tearDown()
			gc.collect()
			torch.cuda.empty_cache()

	@property
	def      __UpperCamelCase							(      self   :						List[Any]):
			UpperCamelCase__				:					Union[str, Any]					=					1
			UpperCamelCase__				:					Union[str, Any]					=					3
			UpperCamelCase__				:					Dict					=					(32, 32)

			UpperCamelCase__				:					int					=					floats_tensor((batch_size, num_channels) + sizes  ,					rng=random.Random(0)).to(UpperCAmelCase_)
			return image

	@property
	def      __UpperCamelCase							(      self   :						Any):
			torch.manual_seed(0)
			UpperCamelCase__				:					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  ,					)
			return model

	@property
	def      __UpperCamelCase							(      self   :						Any):
			torch.manual_seed(0)
			UpperCamelCase__				:					List[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  ,					)
			return model

	@property
	def      __UpperCamelCase							(      self   :						str):
			torch.manual_seed(0)
			UpperCamelCase__				:					Tuple					=					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=1_000  ,					)
			return CLIPTextModel(UpperCAmelCase_)

	@property
	def      __UpperCamelCase							(      self   :						Optional[Any]):
			def extract(*UpperCAmelCase_   :						List[Any]  ,					**UpperCAmelCase_   :						Dict):







					class       __lowercase					:

						def __init__(      self   :						List[Any]):
								UpperCamelCase__				:					Optional[Any]					=					torch.ones([0])





						def      __UpperCamelCase							(      self   :						Dict  ,					UpperCAmelCase_   :						int):
								self.pixel_values.to(UpperCAmelCase_)
								return self


					return Out()

			return extract

	def      __UpperCamelCase							(      self   :						str):
			UpperCamelCase__				:					Optional[Any]					=					'cpu'  # ensure determinism for the device-dependent torch.Generator
			UpperCamelCase__				:					Any					=					self.dummy_cond_unet
			UpperCamelCase__				:					Any					=					DDIMScheduler(
			    beta_start=0.0_00_85  ,					beta_end=0.0_12  ,					beta_schedule='scaled_linear'  ,					clip_sample=UpperCAmelCase_  ,					set_alpha_to_one=UpperCAmelCase_  ,					)

			UpperCamelCase__				:					List[str]					=					self.dummy_vae
			UpperCamelCase__				:					str					=					self.dummy_text_encoder
			UpperCamelCase__				:					Tuple					=					CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip')

			# make sure here that pndm scheduler skips prk
			UpperCamelCase__				:					Optional[Any]					=					StableDiffusionPipeline(
			    unet=UpperCAmelCase_  ,					scheduler=UpperCAmelCase_  ,					vae=UpperCAmelCase_  ,					text_encoder=UpperCAmelCase_  ,					tokenizer=UpperCAmelCase_  ,					safety_checker=UpperCAmelCase_  ,					feature_extractor=self.dummy_extractor  ,					)
			UpperCamelCase__				:					Optional[Any]					=					sd_pipe.to(UpperCAmelCase_)
			sd_pipe.set_progress_bar_config(disable=UpperCAmelCase_)

			UpperCamelCase__				:					Optional[Any]					=					'A painting of a squirrel eating a burger'

			UpperCamelCase__				:					Dict					=					torch.Generator(device=UpperCAmelCase_).manual_seed(0)
			UpperCamelCase__				:					List[Any]					=					sd_pipe([prompt]  ,					generator=UpperCAmelCase_  ,					guidance_scale=6.0  ,					num_inference_steps=2  ,					output_type='np')
			UpperCamelCase__				:					Tuple					=					output.images

			UpperCamelCase__				:					List[Any]					=					torch.Generator(device=UpperCAmelCase_).manual_seed(0)
			UpperCamelCase__				:					Tuple					=					sd_pipe(
			    [prompt]  ,					generator=UpperCAmelCase_  ,					guidance_scale=6.0  ,					num_inference_steps=2  ,					output_type='np'  ,					return_dict=UpperCAmelCase_  ,					)[0]

			UpperCamelCase__				:					List[str]					=					image[0, -3:, -3:, -1]
			UpperCamelCase__				:					Optional[int]					=					image_from_tuple[0, -3:, -3:, -1]

			assert image.shape == (1, 64, 64, 3)
			UpperCamelCase__				:					List[Any]					=					np.array([0.57_56, 0.61_18, 0.50_05, 0.50_41, 0.54_71, 0.47_26, 0.49_76, 0.48_65, 0.48_64])

			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      __UpperCamelCase							(      self   :						Dict):
			UpperCamelCase__				:					List[Any]					=					'cpu'  # ensure determinism for the device-dependent torch.Generator
			UpperCamelCase__				:					int					=					self.dummy_cond_unet
			UpperCamelCase__				:					Dict					=					PNDMScheduler(skip_prk_steps=UpperCAmelCase_)
			UpperCamelCase__				:					Optional[int]					=					self.dummy_vae
			UpperCamelCase__				:					Optional[int]					=					self.dummy_text_encoder
			UpperCamelCase__				:					Union[str, Any]					=					CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip')

			# make sure here that pndm scheduler skips prk
			UpperCamelCase__				:					Dict					=					StableDiffusionPipeline(
			    unet=UpperCAmelCase_  ,					scheduler=UpperCAmelCase_  ,					vae=UpperCAmelCase_  ,					text_encoder=UpperCAmelCase_  ,					tokenizer=UpperCAmelCase_  ,					safety_checker=UpperCAmelCase_  ,					feature_extractor=self.dummy_extractor  ,					)
			UpperCamelCase__				:					Tuple					=					sd_pipe.to(UpperCAmelCase_)
			sd_pipe.set_progress_bar_config(disable=UpperCAmelCase_)

			UpperCamelCase__				:					List[str]					=					'A painting of a squirrel eating a burger'
			UpperCamelCase__				:					Union[str, Any]					=					torch.Generator(device=UpperCAmelCase_).manual_seed(0)
			UpperCamelCase__				:					str					=					sd_pipe([prompt]  ,					generator=UpperCAmelCase_  ,					guidance_scale=6.0  ,					num_inference_steps=2  ,					output_type='np')

			UpperCamelCase__				:					List[str]					=					output.images

			UpperCamelCase__				:					Any					=					torch.Generator(device=UpperCAmelCase_).manual_seed(0)
			UpperCamelCase__				:					Optional[Any]					=					sd_pipe(
			    [prompt]  ,					generator=UpperCAmelCase_  ,					guidance_scale=6.0  ,					num_inference_steps=2  ,					output_type='np'  ,					return_dict=UpperCAmelCase_  ,					)[0]

			UpperCamelCase__				:					Tuple					=					image[0, -3:, -3:, -1]
			UpperCamelCase__				:					Optional[Any]					=					image_from_tuple[0, -3:, -3:, -1]

			assert image.shape == (1, 64, 64, 3)
			UpperCamelCase__				:					List[Any]					=					np.array([0.51_25, 0.57_16, 0.48_28, 0.50_60, 0.56_50, 0.47_68, 0.51_85, 0.48_95, 0.49_93])

			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      __UpperCamelCase							(      self   :						Dict):
			UpperCamelCase__				:					Dict					=					StableDiffusionPipeline.from_pretrained(
			    'hf-internal-testing/tiny-stable-diffusion-lms-pipe'  ,					safety_checker=UpperCAmelCase_)
			assert isinstance(UpperCAmelCase_  ,					UpperCAmelCase_)
			assert isinstance(pipe.scheduler  ,					UpperCAmelCase_)
			assert pipe.safety_checker is None

			UpperCamelCase__				:					List[Any]					=					pipe('example prompt'  ,					num_inference_steps=2).images[0]
			assert image is not None

			# check that there's no error when saving a pipeline with one of the models being None
			with tempfile.TemporaryDirectory() as tmpdirname:
					pipe.save_pretrained(UpperCAmelCase_)
					UpperCamelCase__				:					List[str]					=					StableDiffusionPipeline.from_pretrained(UpperCAmelCase_)

			# sanity check that the pipeline still works
			assert pipe.safety_checker is None
			UpperCamelCase__				:					Optional[Any]					=					pipe('example prompt'  ,					num_inference_steps=2).images[0]
			assert image is not None





	@unittest.skipIf(torch_device != 'cuda'  ,					'This test requires a GPU')
	def      __UpperCamelCase							(      self   :						List[Any]):
			UpperCamelCase__				:					Dict					=					self.dummy_cond_unet
			UpperCamelCase__				:					str					=					PNDMScheduler(skip_prk_steps=UpperCAmelCase_)
			UpperCamelCase__				:					Any					=					self.dummy_vae
			UpperCamelCase__				:					Optional[Any]					=					self.dummy_text_encoder
			UpperCamelCase__				:					str					=					CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip')

			# put models in fp16
			UpperCamelCase__				:					Any					=					unet.half()
			UpperCamelCase__				:					Tuple					=					vae.half()
			UpperCamelCase__				:					Optional[int]					=					bert.half()

			# make sure here that pndm scheduler skips prk
			UpperCamelCase__				:					Optional[int]					=					StableDiffusionPipeline(
			    unet=UpperCAmelCase_  ,					scheduler=UpperCAmelCase_  ,					vae=UpperCAmelCase_  ,					text_encoder=UpperCAmelCase_  ,					tokenizer=UpperCAmelCase_  ,					safety_checker=UpperCAmelCase_  ,					feature_extractor=self.dummy_extractor  ,					)
			UpperCamelCase__				:					Dict					=					sd_pipe.to(UpperCAmelCase_)
			sd_pipe.set_progress_bar_config(disable=UpperCAmelCase_)

			UpperCamelCase__				:					Any					=					'A painting of a squirrel eating a burger'
			UpperCamelCase__				:					int					=					sd_pipe([prompt]  ,					num_inference_steps=2  ,					output_type='np').images

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







@nightly
@require_torch_gpu
class       __lowercase					(unittest.TestCase    ):

	def      __UpperCamelCase							(      self   :						Optional[int]):
			# clean up the VRAM after each test
			super().tearDown()
			gc.collect()
			torch.cuda.empty_cache()

	def      __UpperCamelCase							(      self   :						List[Any]):
			UpperCamelCase__				:					Optional[int]					=					StableDiffusionPipeline.from_pretrained('runwayml/stable-diffusion-v1-5'  ,					safety_checker=UpperCAmelCase_)
			UpperCamelCase__				:					Union[str, Any]					=					LMSDiscreteScheduler.from_config(sd_pipe.scheduler.config)
			UpperCamelCase__				:					Optional[Any]					=					sd_pipe.to(UpperCAmelCase_)
			sd_pipe.set_progress_bar_config(disable=UpperCAmelCase_)

			UpperCamelCase__				:					List[Any]					=					(
			    'portrait of girl with smokey eyes makeup in abandoned hotel, grange clothes, redshift, wide high angle'
			    ' coloured polaroid photograph with flash, kodak film, hyper real, stunning moody cinematography, with'
			    ' anamorphic lenses, by maripol, fallen angels by wong kar - wai, style of suspiria and neon demon and'
			    ' children from bahnhof zoo, detailed '
			)
			UpperCamelCase__				:					Any					=					4_003_660_346
			UpperCamelCase__				:					Any					=					7

			# without safety guidance (sld_guidance_scale = 0)
			UpperCamelCase__				:					int					=					torch.manual_seed(UpperCAmelCase_)
			UpperCamelCase__				:					Optional[int]					=					sd_pipe(
			    [prompt]  ,					generator=UpperCAmelCase_  ,					guidance_scale=UpperCAmelCase_  ,					num_inference_steps=50  ,					output_type='np'  ,					width=512  ,					height=512  ,					sld_guidance_scale=0  ,					)

			UpperCamelCase__				:					str					=					output.images
			UpperCamelCase__				:					Union[str, Any]					=					image[0, -3:, -3:, -1]
			UpperCamelCase__				:					Tuple					=					[0.22_78, 0.22_31, 0.22_49, 0.23_33, 0.23_03, 0.18_85, 0.22_73, 0.21_44, 0.21_76]

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

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

			# without safety guidance (strong configuration)
			UpperCamelCase__				:					Tuple					=					torch.manual_seed(UpperCAmelCase_)
			UpperCamelCase__				:					str					=					sd_pipe(
			    [prompt]  ,					generator=UpperCAmelCase_  ,					guidance_scale=UpperCAmelCase_  ,					num_inference_steps=50  ,					output_type='np'  ,					width=512  ,					height=512  ,					sld_guidance_scale=2_000  ,					sld_warmup_steps=7  ,					sld_threshold=0.0_25  ,					sld_momentum_scale=0.5  ,					sld_mom_beta=0.7  ,					)

			UpperCamelCase__				:					Dict					=					output.images
			UpperCamelCase__				:					str					=					image[0, -3:, -3:, -1]
			UpperCamelCase__				:					Tuple					=					[0.23_83, 0.22_76, 0.2_36, 0.21_92, 0.21_86, 0.20_53, 0.19_71, 0.19_01, 0.17_19]

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

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

	def      __UpperCamelCase							(      self   :						Optional[Any]):
			UpperCamelCase__				:					Dict					=					StableDiffusionPipeline.from_pretrained('runwayml/stable-diffusion-v1-5'  ,					safety_checker=UpperCAmelCase_)
			UpperCamelCase__				:					str					=					LMSDiscreteScheduler.from_config(sd_pipe.scheduler.config)
			UpperCamelCase__				:					Dict					=					sd_pipe.to(UpperCAmelCase_)
			sd_pipe.set_progress_bar_config(disable=UpperCAmelCase_)

			UpperCamelCase__				:					str					=					'padme amidala taking a bath artwork, safe for work, no nudity'
			UpperCamelCase__				:					Tuple					=					2_734_971_755
			UpperCamelCase__				:					Tuple					=					7

			UpperCamelCase__				:					Tuple					=					torch.manual_seed(UpperCAmelCase_)
			UpperCamelCase__				:					int					=					sd_pipe(
			    [prompt]  ,					generator=UpperCAmelCase_  ,					guidance_scale=UpperCAmelCase_  ,					num_inference_steps=50  ,					output_type='np'  ,					width=512  ,					height=512  ,					sld_guidance_scale=0  ,					)

			UpperCamelCase__				:					int					=					output.images
			UpperCamelCase__				:					Union[str, Any]					=					image[0, -3:, -3:, -1]
			UpperCamelCase__				:					Any					=					[0.35_02, 0.36_22, 0.33_96, 0.36_42, 0.34_78, 0.33_18, 0.35, 0.33_48, 0.32_97]

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

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

			UpperCamelCase__				:					List[str]					=					torch.manual_seed(UpperCAmelCase_)
			UpperCamelCase__				:					Union[str, Any]					=					sd_pipe(
			    [prompt]  ,					generator=UpperCAmelCase_  ,					guidance_scale=UpperCAmelCase_  ,					num_inference_steps=50  ,					output_type='np'  ,					width=512  ,					height=512  ,					sld_guidance_scale=2_000  ,					sld_warmup_steps=7  ,					sld_threshold=0.0_25  ,					sld_momentum_scale=0.5  ,					sld_mom_beta=0.7  ,					)

			UpperCamelCase__				:					Tuple					=					output.images
			UpperCamelCase__				:					List[str]					=					image[0, -3:, -3:, -1]
			UpperCamelCase__				:					Union[str, Any]					=					[0.55_31, 0.52_06, 0.48_95, 0.51_56, 0.51_82, 0.47_51, 0.48_02, 0.48_03, 0.44_43]

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

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





	def      __UpperCamelCase							(      self   :						Any):
			UpperCamelCase__				:					Optional[Any]					=					StableDiffusionPipeline.from_pretrained('runwayml/stable-diffusion-v1-5')
			UpperCamelCase__				:					Optional[Any]					=					sd_pipe.to(UpperCAmelCase_)
			sd_pipe.set_progress_bar_config(disable=UpperCAmelCase_)

			UpperCamelCase__				:					int					=					(
			    'the four horsewomen of the apocalypse, painting by tom of finland, gaston bussiere, craig mullins, j. c.'
			    ' leyendecker'
			)
			UpperCamelCase__				:					Any					=					1_044_355_234
			UpperCamelCase__				:					Optional[int]					=					12

			UpperCamelCase__				:					Optional[int]					=					torch.manual_seed(UpperCAmelCase_)
			UpperCamelCase__				:					str					=					sd_pipe(
			    [prompt]  ,					generator=UpperCAmelCase_  ,					guidance_scale=UpperCAmelCase_  ,					num_inference_steps=50  ,					output_type='np'  ,					width=512  ,					height=512  ,					sld_guidance_scale=0  ,					)

			UpperCamelCase__				:					List[str]					=					output.images
			UpperCamelCase__				:					Any					=					image[0, -3:, -3:, -1]
			UpperCamelCase__				:					str					=					np.array([0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0])

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

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

			UpperCamelCase__				:					int					=					torch.manual_seed(UpperCAmelCase_)
			UpperCamelCase__				:					List[str]					=					sd_pipe(
			    [prompt]  ,					generator=UpperCAmelCase_  ,					guidance_scale=UpperCAmelCase_  ,					num_inference_steps=50  ,					output_type='np'  ,					width=512  ,					height=512  ,					sld_guidance_scale=2_000  ,					sld_warmup_steps=7  ,					sld_threshold=0.0_25  ,					sld_momentum_scale=0.5  ,					sld_mom_beta=0.7  ,					)

			UpperCamelCase__				:					Optional[Any]					=					output.images
			UpperCamelCase__				:					List[Any]					=					image[0, -3:, -3:, -1]
			UpperCamelCase__				:					str					=					np.array([0.58_18, 0.62_85, 0.68_35, 0.60_19, 0.6_25, 0.67_54, 0.60_96, 0.63_34, 0.65_61])
			assert image.shape == (1, 512, 512, 3)

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




