






import argparse
import os

import evaluate
import torch
from datasets import load_dataset
from torch.optim import AdamW
from torch.utils.data import DataLoader
from transformers import AutoModelForSequenceClassification, AutoTokenizer, get_linear_schedule_with_warmup, set_seed

from accelerate import Accelerator, DistributedType


########################################################################
# This is a fully working simple example to use Accelerate,
# specifically showcasing the experiment tracking capability,
# and builds off the `nlp_example.py` script.
#
# This example trains a Bert base model on GLUE MRPC
# in any of the following settings (with the same script):
#   - single CPU or single GPU
#   - multi GPUS (using PyTorch distributed mode)
#   - (multi) TPUs
#   - fp16 (mixed-precision) or fp32 (normal precision)
#
# To help focus on the differences in the code, building `DataLoaders`
# was refactored into its own function.
# New additions from the base script can be found quickly by
# looking for the # New Code # tags
#
# To run it in each of these various modes, follow the instructions
# in the readme for examples:
# https://github.com/huggingface/accelerate/tree/main/examples
#
########################################################################

__UpperCamelCase      :      Optional[Any]     						=						16
__UpperCamelCase      :      Tuple     						=						32





def 				snake_case  (     lowerCamelCase				,  lowerCamelCase = 16					):




 '''simple docstring'''







 __lowercase             =  AutoTokenizer.from_pretrained("""bert-base-cased"""					)
 __lowercase             =  load_dataset("""glue"""				,  """mrpc"""					)

 def tokenize_function(lowerCamelCase					):
  # max_length=None => use the model max length (it's actually the default)
  __lowercase             =  tokenizer(examples["""sentence1"""]				,  examples["""sentence2"""]				,  truncation=lowerCamelCase				,  max_length=lowerCamelCase					)
  return outputs

 # Apply the method we just defined to all the examples in all the splits of the dataset
 # starting with the main process first:
 with accelerator.main_process_first():
  __lowercase             =  datasets.map(
      lowerCamelCase				,  batched=lowerCamelCase				,  remove_columns=["""idx""", """sentence1""", """sentence2"""]				,  )

 # We also rename the 'label' column to 'labels' which is the expected name for labels by the models of the
 # transformers library
 __lowercase             =  tokenized_datasets.rename_column("""label"""				,  """labels"""					)

 def collate_fn(lowerCamelCase					):
  # On TPU it's best to pad everything to the same length or training will be very slow.
  __lowercase             =  128 if accelerator.distributed_type == DistributedType.TPU else None
  # When using mixed precision we want round multiples of 8/16
  if accelerator.mixed_precision == "fp8":
   __lowercase             =  16
  elif accelerator.mixed_precision != "no":
   __lowercase             =  8
  else:
   __lowercase             =  None

  return tokenizer.pad(
      lowerCamelCase				,  padding="""longest"""				,  max_length=lowerCamelCase				,  pad_to_multiple_of=lowerCamelCase				,  return_tensors="""pt"""				,  )

 # Instantiate dataloaders.
 __lowercase             =  DataLoader(
     tokenized_datasets["""train"""]				,  shuffle=lowerCamelCase				,  collate_fn=lowerCamelCase				,  batch_size=lowerCamelCase					)
 __lowercase             =  DataLoader(
     tokenized_datasets["""validation"""]				,  shuffle=lowerCamelCase				,  collate_fn=lowerCamelCase				,  batch_size=lowerCamelCase					)

 return train_dataloader, eval_dataloader


# For testing only
if os.environ.get("""TESTING_MOCKED_DATALOADERS""", None) == "1":
      from accelerate.test_utils.training import mocked_dataloaders

      __UpperCamelCase      :      Optional[int]     						=						mocked_dataloaders  # noqa: F811





def 				snake_case  (     lowerCamelCase				,  lowerCamelCase					):




 '''simple docstring'''







 if os.environ.get("""TESTING_MOCKED_DATALOADERS"""				,  lowerCamelCase					) == "1":
  __lowercase             =  2
 # Initialize Accelerator

 # New Code #
 # We pass in "all" to `log_with` to grab all available trackers in the environment
 # Note: If using a custom `Tracker` class, should be passed in here such as:
 # >>> log_with = ["all", MyCustomTrackerClassInstance()]
 if args.with_tracking:
  __lowercase             =  Accelerator(
      cpu=args.cpu				,  mixed_precision=args.mixed_precision				,  log_with="""all"""				,  project_dir=args.project_dir					)
 else:
  __lowercase             =  Accelerator(cpu=args.cpu				,  mixed_precision=args.mixed_precision					)
 # Sample hyper-parameters for learning rate, batch size, seed and a few other HPs
 __lowercase             =  config["""lr"""]
 __lowercase             =  int(config["""num_epochs"""]					)
 __lowercase             =  int(config["""seed"""]					)
 __lowercase             =  int(config["""batch_size"""]					)
 set_seed(lowerCamelCase					)

 __lowercase       ,						__lowercase             =  get_dataloaders(lowerCamelCase				,  lowerCamelCase					)
 __lowercase             =  evaluate.load("""glue"""				,  """mrpc"""					)

 # If the batch size is too big we use gradient accumulation
 __lowercase             =  1
 if batch_size > MAX_GPU_BATCH_SIZE and accelerator.distributed_type != DistributedType.TPU:
  __lowercase             =  batch_size // MAX_GPU_BATCH_SIZE
  __lowercase             =  MAX_GPU_BATCH_SIZE

 # Instantiate the model (we build the model here so that the seed also control new weights initialization)
 __lowercase             =  AutoModelForSequenceClassification.from_pretrained("""bert-base-cased"""				,  return_dict=lowerCamelCase					)

 # We could avoid this line since the accelerator is set with `device_placement=True` (default value).
 # Note that if you are placing tensors on devices manually, this line absolutely needs to be before the optimizer
 # creation otherwise training will not work on TPU (`accelerate` will kindly throw an error to make us aware of that).
 __lowercase             =  model.to(accelerator.device					)

 # Instantiate optimizer
 __lowercase             =  AdamW(params=model.parameters()				,  lr=lowerCamelCase					)

 # Instantiate scheduler
 __lowercase             =  get_linear_schedule_with_warmup(
     optimizer=lowerCamelCase				,  num_warmup_steps=100				,  num_training_steps=(len(lowerCamelCase					) * num_epochs) // gradient_accumulation_steps				,  )

 # Prepare everything
 # There is no specific order to remember, we just need to unpack the objects in the same order we gave them to the
 # prepare method.
 __lowercase       ,						__lowercase       ,						__lowercase       ,						__lowercase       ,						__lowercase             =  accelerator.prepare(
     lowerCamelCase				,  lowerCamelCase				,  lowerCamelCase				,  lowerCamelCase				,  lowerCamelCase					)

 # New Code #
 # We need to initialize the trackers we use. Overall configurations can also be stored
 if args.with_tracking:
  __lowercase             =  os.path.split(lowerCamelCase					)[-1].split("""."""					)[0]
  accelerator.init_trackers(lowerCamelCase				,  lowerCamelCase					)

 # Now we train the model
 for epoch in range(lowerCamelCase					):
  model.train()
  # New Code #
  # For our tracking example, we will log the total loss of each epoch
  if args.with_tracking:
   __lowercase             =  0
  for step, batch in enumerate(lowerCamelCase					):
   # We could avoid this line since we set the accelerator with `device_placement=True`.
   batch.to(accelerator.device					)
   __lowercase             =  model(**lowerCamelCase					)
   __lowercase             =  outputs.loss
   # New Code #
   if args.with_tracking:
    total_loss += loss.detach().float()
   __lowercase             =  loss / gradient_accumulation_steps
   accelerator.backward(lowerCamelCase					)
   if step % gradient_accumulation_steps == 0:
    optimizer.step()
    lr_scheduler.step()
    optimizer.zero_grad()

  model.eval()
  for step, batch in enumerate(lowerCamelCase					):
   # We could avoid this line since we set the accelerator with `device_placement=True` (the default).
   batch.to(accelerator.device					)
   with torch.no_grad():
    __lowercase             =  model(**lowerCamelCase					)
   __lowercase             =  outputs.logits.argmax(dim=-1					)
   __lowercase       ,						__lowercase             =  accelerator.gather_for_metrics((predictions, batch["""labels"""])					)
   metric.add_batch(
       predictions=lowerCamelCase				,  references=lowerCamelCase				,  )

  __lowercase             =  metric.compute()
  # Use accelerator.print to print only on the main process.
  accelerator.print(F'epoch {epoch}:'				,  lowerCamelCase					)

  # New Code #
  # To actually log, we call `Accelerator.log`
  # The values passed can be of `str`, `int`, `float` or `dict` of `str` to `float`/`int`
  if args.with_tracking:
   accelerator.log(
       {
           """accuracy""": eval_metric["""accuracy"""],
           """f1""": eval_metric["""f1"""],
           """train_loss""": total_loss.item() / len(lowerCamelCase					),
           """epoch""": epoch,
       }				,  step=lowerCamelCase				,  )

    # New Code #
    # When a run is finished, you should call `accelerator.end_training()`
    # to close all of the open trackers
 if args.with_tracking:
  accelerator.end_training()





def 				snake_case  (     ):




 '''simple docstring'''







 __lowercase             =  argparse.ArgumentParser(description="""Simple example of training script."""					)
 parser.add_argument(
     """--mixed_precision"""				,  type=lowerCamelCase				,  default=lowerCamelCase				,  choices=["""no""", """fp16""", """bf16""", """fp8"""]				,  help="""Whether to use mixed precision. Choose"""
     """between fp16 and bf16 (bfloat16). Bf16 requires PyTorch >= 1.10."""
     """and an Nvidia Ampere GPU."""				,  )
 parser.add_argument("""--cpu"""				,  action="""store_true"""				,  help="""If passed, will train on the CPU."""					)
 parser.add_argument(
     """--with_tracking"""				,  action="""store_true"""				,  help="""Whether to load in all available experiment trackers from the environment and use them for logging."""				,  )
 parser.add_argument(
     """--project_dir"""				,  type=lowerCamelCase				,  default="""logs"""				,  help="""Location on where to store experiment tracking logs` and relevent project information"""				,  )
 __lowercase             =  parser.parse_args()
 __lowercase             =  {"""lr""": 2e-5, """num_epochs""": 3, """seed""": 42, """batch_size""": 16}
 training_function(lowerCamelCase				,  lowerCamelCase					)


if __name__ == "__main__":
      main()









import tempfile
import unittest

import numpy as np

from diffusers import (
    DDIMScheduler,
    DPMSolverMultistepScheduler,
    EulerAncestralDiscreteScheduler,
    EulerDiscreteScheduler,
    LMSDiscreteScheduler,
    OnnxStableDiffusionPipeline,
    PNDMScheduler,
)
from diffusers.utils.testing_utils import is_onnx_available, nightly, require_onnxruntime, require_torch_gpu

from ..test_pipelines_onnx_common import OnnxPipelineTesterMixin


if is_onnx_available():
						import onnxruntime as ort


class        __UpperCamelCase    (     _lowerCAmelCase   ,			unittest.TestCase   ):
	__snake_case	:Union[str, Any]   =       'hf-internal-testing/tiny-random-OnnxStableDiffusionPipeline'







	def  _a   (  self      :							Any       ,				_lowerCAmelCase      :							str=0			)		-> str:
		"""simple docstring"""







		__lowercase             =  np.random.RandomState(_lowerCAmelCase			)
		__lowercase             =  {
		    """prompt""": """A painting of a squirrel eating a burger""",
		    """generator""": generator,
		    """num_inference_steps""": 2,
		    """guidance_scale""": 7.5,
		    """output_type""": """numpy""",
		}
		return inputs







	def  _a   (  self      :							int			)		-> List[Any]:
		"""simple docstring"""







		__lowercase             =  OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint       ,				provider="""CPUExecutionProvider"""			)
		pipe.set_progress_bar_config(disable=_lowerCAmelCase			)

		__lowercase             =  self.get_dummy_inputs()
		__lowercase             =  pipe(**_lowerCAmelCase			).images
		__lowercase             =  image[0, -3:, -3:, -1]

		assert image.shape == (1, 128, 128, 3)
		__lowercase             =  np.array([0.65_072, 0.58_492, 0.48_219, 0.55_521, 0.53_180, 0.55_939, 0.50_697, 0.39_800, 0.46_455]			)

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







	def  _a   (  self      :							Union[str, Any]			)		-> str:
		"""simple docstring"""







		__lowercase             =  OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint       ,				provider="""CPUExecutionProvider"""			)
		__lowercase             =  PNDMScheduler.from_config(pipe.scheduler.config       ,				skip_prk_steps=_lowerCAmelCase			)
		pipe.set_progress_bar_config(disable=_lowerCAmelCase			)

		__lowercase             =  self.get_dummy_inputs()
		__lowercase             =  pipe(**_lowerCAmelCase			).images
		__lowercase             =  image[0, -3:, -3:, -1]

		assert image.shape == (1, 128, 128, 3)
		__lowercase             =  np.array([0.65_863, 0.59_425, 0.49_326, 0.56_313, 0.53_875, 0.56_627, 0.51_065, 0.39_777, 0.46_330]			)

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







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







		__lowercase             =  OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint       ,				provider="""CPUExecutionProvider"""			)
		__lowercase             =  LMSDiscreteScheduler.from_config(pipe.scheduler.config			)
		pipe.set_progress_bar_config(disable=_lowerCAmelCase			)

		__lowercase             =  self.get_dummy_inputs()
		__lowercase             =  pipe(**_lowerCAmelCase			).images
		__lowercase             =  image[0, -3:, -3:, -1]

		assert image.shape == (1, 128, 128, 3)
		__lowercase             =  np.array([0.53_755, 0.60_786, 0.47_402, 0.49_488, 0.51_869, 0.49_819, 0.47_985, 0.38_957, 0.44_279]			)

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







	def  _a   (  self      :							Tuple			)		-> int:
		"""simple docstring"""







		__lowercase             =  OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint       ,				provider="""CPUExecutionProvider"""			)
		__lowercase             =  EulerDiscreteScheduler.from_config(pipe.scheduler.config			)
		pipe.set_progress_bar_config(disable=_lowerCAmelCase			)

		__lowercase             =  self.get_dummy_inputs()
		__lowercase             =  pipe(**_lowerCAmelCase			).images
		__lowercase             =  image[0, -3:, -3:, -1]

		assert image.shape == (1, 128, 128, 3)
		__lowercase             =  np.array([0.53_755, 0.60_786, 0.47_402, 0.49_488, 0.51_869, 0.49_819, 0.47_985, 0.38_957, 0.44_279]			)

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







	def  _a   (  self      :							Tuple			)		-> Union[str, Any]:
		"""simple docstring"""







		__lowercase             =  OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint       ,				provider="""CPUExecutionProvider"""			)
		__lowercase             =  EulerAncestralDiscreteScheduler.from_config(pipe.scheduler.config			)
		pipe.set_progress_bar_config(disable=_lowerCAmelCase			)

		__lowercase             =  self.get_dummy_inputs()
		__lowercase             =  pipe(**_lowerCAmelCase			).images
		__lowercase             =  image[0, -3:, -3:, -1]

		assert image.shape == (1, 128, 128, 3)
		__lowercase             =  np.array([0.53_817, 0.60_812, 0.47_384, 0.49_530, 0.51_894, 0.49_814, 0.47_984, 0.38_958, 0.44_271]			)

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







	def  _a   (  self      :							Union[str, Any]			)		-> List[str]:
		"""simple docstring"""







		__lowercase             =  OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint       ,				provider="""CPUExecutionProvider"""			)
		__lowercase             =  DPMSolverMultistepScheduler.from_config(pipe.scheduler.config			)
		pipe.set_progress_bar_config(disable=_lowerCAmelCase			)

		__lowercase             =  self.get_dummy_inputs()
		__lowercase             =  pipe(**_lowerCAmelCase			).images
		__lowercase             =  image[0, -3:, -3:, -1]

		assert image.shape == (1, 128, 128, 3)
		__lowercase             =  np.array([0.53_895, 0.60_808, 0.47_933, 0.49_608, 0.51_886, 0.49_950, 0.48_053, 0.38_957, 0.44_200]			)

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







	def  _a   (  self      :							List[str]			)		-> List[str]:
		"""simple docstring"""







		__lowercase             =  OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint       ,				provider="""CPUExecutionProvider"""			)
		pipe.set_progress_bar_config(disable=_lowerCAmelCase			)

		__lowercase             =  self.get_dummy_inputs()
		__lowercase             =  3 * [inputs["""prompt"""]]

		# forward
		__lowercase             =  pipe(**_lowerCAmelCase			)
		__lowercase             =  output.images[0, -3:, -3:, -1]

		__lowercase             =  self.get_dummy_inputs()
		__lowercase             =  3 * [inputs.pop("""prompt"""			)]

		__lowercase             =  pipe.tokenizer(
		    _lowerCAmelCase       ,				padding="""max_length"""       ,				max_length=pipe.tokenizer.model_max_length       ,				truncation=_lowerCAmelCase       ,				return_tensors="""np"""       ,				)
		__lowercase             =  text_inputs["""input_ids"""]

		__lowercase             =  pipe.text_encoder(input_ids=text_inputs.astype(np.intaa			)			)[0]

		__lowercase             =  prompt_embeds

		# forward
		__lowercase             =  pipe(**_lowerCAmelCase			)
		__lowercase             =  output.images[0, -3:, -3:, -1]

		assert np.abs(image_slice_a.flatten() - image_slice_a.flatten()			).max() < 1e-4







	def  _a   (  self      :							int			)		-> str:
		"""simple docstring"""







		__lowercase             =  OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint       ,				provider="""CPUExecutionProvider"""			)
		pipe.set_progress_bar_config(disable=_lowerCAmelCase			)

		__lowercase             =  self.get_dummy_inputs()
		__lowercase             =  3 * ["""this is a negative prompt"""]
		__lowercase             =  negative_prompt
		__lowercase             =  3 * [inputs["""prompt"""]]

		# forward
		__lowercase             =  pipe(**_lowerCAmelCase			)
		__lowercase             =  output.images[0, -3:, -3:, -1]

		__lowercase             =  self.get_dummy_inputs()
		__lowercase             =  3 * [inputs.pop("""prompt"""			)]

		__lowercase             =  []
		for p in [prompt, negative_prompt]:
			__lowercase             =  pipe.tokenizer(
			    _lowerCAmelCase       ,				padding="""max_length"""       ,				max_length=pipe.tokenizer.model_max_length       ,				truncation=_lowerCAmelCase       ,				return_tensors="""np"""       ,				)
			__lowercase             =  text_inputs["""input_ids"""]

			embeds.append(pipe.text_encoder(input_ids=text_inputs.astype(np.intaa			)			)[0]			)

		__lowercase       ,						__lowercase             =  embeds

		# forward
		__lowercase             =  pipe(**_lowerCAmelCase			)
		__lowercase             =  output.images[0, -3:, -3:, -1]

		assert np.abs(image_slice_a.flatten() - image_slice_a.flatten()			).max() < 1e-4


@nightly
@require_onnxruntime
@require_torch_gpu
class        __UpperCamelCase    (     unittest.TestCase   ):







	@property
	def  _a   (  self      :							Dict			)		-> str:
		"""simple docstring"""







		return (
		    "CUDAExecutionProvider",
		    {
		        "gpu_mem_limit": "15000000000",  # 15GB
		        "arena_extend_strategy": "kSameAsRequested",
		    },
		)







	@property
	def  _a   (  self      :							Optional[Any]			)		-> Optional[Any]:
		"""simple docstring"""







		__lowercase             =  ort.SessionOptions()
		__lowercase             =  False
		return options







	def  _a   (  self      :							List[Any]			)		-> Union[str, Any]:
		"""simple docstring"""







		__lowercase             =  OnnxStableDiffusionPipeline.from_pretrained(
		    """CompVis/stable-diffusion-v1-4"""       ,				revision="""onnx"""       ,				safety_checker=_lowerCAmelCase       ,				feature_extractor=_lowerCAmelCase       ,				provider=self.gpu_provider       ,				sess_options=self.gpu_options       ,				)
		sd_pipe.set_progress_bar_config(disable=_lowerCAmelCase			)

		__lowercase             =  """A painting of a squirrel eating a burger"""
		np.random.seed(0			)
		__lowercase             =  sd_pipe([prompt]       ,				guidance_scale=6.0       ,				num_inference_steps=10       ,				output_type="""np"""			)
		__lowercase             =  output.images

		__lowercase             =  image[0, -3:, -3:, -1]

		assert image.shape == (1, 512, 512, 3)
		__lowercase             =  np.array([0.0_452, 0.0_390, 0.0_087, 0.0_350, 0.0_617, 0.0_364, 0.0_544, 0.0_523, 0.0_720]			)

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







	def  _a   (  self      :							Tuple			)		-> Any:
		"""simple docstring"""







		__lowercase             =  DDIMScheduler.from_pretrained(
		    """runwayml/stable-diffusion-v1-5"""       ,				subfolder="""scheduler"""       ,				revision="""onnx"""			)
		__lowercase             =  OnnxStableDiffusionPipeline.from_pretrained(
		    """runwayml/stable-diffusion-v1-5"""       ,				revision="""onnx"""       ,				scheduler=_lowerCAmelCase       ,				safety_checker=_lowerCAmelCase       ,				feature_extractor=_lowerCAmelCase       ,				provider=self.gpu_provider       ,				sess_options=self.gpu_options       ,				)
		sd_pipe.set_progress_bar_config(disable=_lowerCAmelCase			)

		__lowercase             =  """open neural network exchange"""
		__lowercase             =  np.random.RandomState(0			)
		__lowercase             =  sd_pipe([prompt]       ,				guidance_scale=7.5       ,				num_inference_steps=10       ,				generator=_lowerCAmelCase       ,				output_type="""np"""			)
		__lowercase             =  output.images
		__lowercase             =  image[0, -3:, -3:, -1]

		assert image.shape == (1, 512, 512, 3)
		__lowercase             =  np.array([0.2_867, 0.1_974, 0.1_481, 0.7_294, 0.7_251, 0.6_667, 0.4_194, 0.5_642, 0.6_486]			)

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







	def  _a   (  self      :							Dict			)		-> Dict:
		"""simple docstring"""







		__lowercase             =  LMSDiscreteScheduler.from_pretrained(
		    """runwayml/stable-diffusion-v1-5"""       ,				subfolder="""scheduler"""       ,				revision="""onnx"""			)
		__lowercase             =  OnnxStableDiffusionPipeline.from_pretrained(
		    """runwayml/stable-diffusion-v1-5"""       ,				revision="""onnx"""       ,				scheduler=_lowerCAmelCase       ,				safety_checker=_lowerCAmelCase       ,				feature_extractor=_lowerCAmelCase       ,				provider=self.gpu_provider       ,				sess_options=self.gpu_options       ,				)
		sd_pipe.set_progress_bar_config(disable=_lowerCAmelCase			)

		__lowercase             =  """open neural network exchange"""
		__lowercase             =  np.random.RandomState(0			)
		__lowercase             =  sd_pipe([prompt]       ,				guidance_scale=7.5       ,				num_inference_steps=10       ,				generator=_lowerCAmelCase       ,				output_type="""np"""			)
		__lowercase             =  output.images
		__lowercase             =  image[0, -3:, -3:, -1]

		assert image.shape == (1, 512, 512, 3)
		__lowercase             =  np.array([0.2_306, 0.1_959, 0.1_593, 0.6_549, 0.6_394, 0.5_408, 0.5_065, 0.6_010, 0.6_161]			)

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







	def  _a   (  self      :							str			)		-> List[str]:
		"""simple docstring"""







		__lowercase             =  0

		def test_callback_fn(_lowerCAmelCase      :							int       ,				_lowerCAmelCase      :							int       ,				_lowerCAmelCase      :							np.ndarray			) -> None:
			__lowercase             =  True
			nonlocal number_of_steps
			number_of_steps += 1
			if step == 0:
				assert latents.shape == (1, 4, 64, 64)
				__lowercase             =  latents[0, -3:, -3:, -1]
				__lowercase             =  np.array(
				    [-0.6_772, -0.3_835, -1.2_456, 0.1_905, -1.0_974, 0.6_967, -1.9_353, 0.0_178, 1.0_167]			)

				assert np.abs(latents_slice.flatten() - expected_slice			).max() < 1e-3
			elif step == 5:
				assert latents.shape == (1, 4, 64, 64)
				__lowercase             =  latents[0, -3:, -3:, -1]
				__lowercase             =  np.array(
				    [-0.3_351, 0.2_241, -0.1_837, -0.2_325, -0.6_577, 0.3_393, -0.0_241, 0.5_899, 1.3_875]			)

				assert np.abs(latents_slice.flatten() - expected_slice			).max() < 1e-3

		__lowercase             =  False

		__lowercase             =  OnnxStableDiffusionPipeline.from_pretrained(
		    """runwayml/stable-diffusion-v1-5"""       ,				revision="""onnx"""       ,				safety_checker=_lowerCAmelCase       ,				feature_extractor=_lowerCAmelCase       ,				provider=self.gpu_provider       ,				sess_options=self.gpu_options       ,				)
		pipe.set_progress_bar_config(disable=_lowerCAmelCase			)

		__lowercase             =  """Andromeda galaxy in a bottle"""

		__lowercase             =  np.random.RandomState(0			)
		pipe(
		    prompt=_lowerCAmelCase       ,				num_inference_steps=5       ,				guidance_scale=7.5       ,				generator=_lowerCAmelCase       ,				callback=_lowerCAmelCase       ,				callback_steps=1       ,				)
		assert test_callback_fn.has_been_called
		assert number_of_steps == 6







	def  _a   (  self      :							Union[str, Any]			)		-> Tuple:
		"""simple docstring"""







		__lowercase             =  OnnxStableDiffusionPipeline.from_pretrained(
		    """runwayml/stable-diffusion-v1-5"""       ,				revision="""onnx"""       ,				safety_checker=_lowerCAmelCase       ,				feature_extractor=_lowerCAmelCase       ,				provider=self.gpu_provider       ,				sess_options=self.gpu_options       ,				)
		assert isinstance(_lowerCAmelCase       ,				_lowerCAmelCase			)
		assert pipe.safety_checker is None

		__lowercase             =  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(_lowerCAmelCase			)
			__lowercase             =  OnnxStableDiffusionPipeline.from_pretrained(_lowerCAmelCase			)

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

