


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


SCREAMING_SNAKE_CASE			: Optional[Any]				   =					logging.getLogger(__name__)

SCREAMING_SNAKE_CASE			: Dict				   =					"pytorch_model.bin"






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






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






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



def 							UpperCamelCase_( lowerCamelCase_				,					lowerCamelCase_				,					lowerCamelCase_				,					lowerCamelCase_				,					lowerCamelCase_				,					lowerCamelCase_					)       ->		Dict:
    _lowercase			:				int 						=						datasets.concatenate_datasets([infer_input, infer_output]				,					axis=1					)

    if args.do_filter_by_confidence:
        _lowercase			:				int 						=						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
        _lowercase			:				Tuple 						=						int(eval_result * len(lowerCamelCase_					)					)
        print(lowerCamelCase_					)
        _lowercase			:				str 						=						dataset.sort('probability'				,					reverse=lowerCamelCase_					)
        _lowercase			:				List[Any] 						=						dataset.select(range(lowerCamelCase_					)					)

    _lowercase			:				Dict 						=						dataset.remove_columns(['label', 'probability']					)
    _lowercase			:				Optional[int] 						=						dataset.rename_column('prediction'				,					'label'					)
    _lowercase			:				Any 						=						dataset.map(lambda lowerCamelCase_					: {"label": idalabel[example["label"]]}					)
    _lowercase			:				int 						=						dataset.shuffle(seed=args.seed					)

    _lowercase			:				Dict 						=						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 							UpperCamelCase_( lowerCamelCase_				,					lowerCamelCase_				,					lowerCamelCase_				,					lowerCamelCase_				,					**lowerCamelCase_					)       ->		Any:
    _lowercase			:				List[Any] 						=						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()

    _lowercase			:				Any 						=						STModelArguments(model_name_or_path=lowerCamelCase_					)
    _lowercase			:				List[str] 						=						STDataArguments(train_file=lowerCamelCase_				,					infer_file=lowerCamelCase_					)
    _lowercase			:				Union[str, Any] 						=						STTrainingArguments(output_dir=lowerCamelCase_					)
    _lowercase			:				str 						=						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
    _lowercase			:				str 						=						{}
    _lowercase			:				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
    _lowercase			:				Optional[int] 						=						args.train_file
    _lowercase			:				str 						=						args.infer_file

    if args.evaluation_strategy != IntervalStrategy.NO.value:
        assert args.eval_file is not None
        _lowercase			:				List[Any] 						=						args.eval_file

    for key in data_files:
        _lowercase			:				Union[str, Any] 						=						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:
            _lowercase			:				Tuple 						=						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...'					)
    _lowercase			:				int 						=						F'''{args.output_dir}/self-train_iter-{{}}'''.format
    _lowercase			:				str 						=						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()

    _lowercase			:				Optional[Any] 						=						None
    _lowercase			:				Tuple 						=						None
    _lowercase			:				List[str] 						=						0
    _lowercase			:				str 						=						False
    # Show the progress bar
    _lowercase			:				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					)					):
        _lowercase			:				Dict 						=						data_dir_format(lowerCamelCase_					)
        assert os.path.exists(lowerCamelCase_					)

        # Stage 1: initial fine-tuning for iteration = 0 or pseudo-training for
        # iteration > 0
        _lowercase			:				Optional[int] 						=						os.path.join(lowerCamelCase_				,					'stage-1'					)
        _lowercase			:				Union[str, 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}					)

        _lowercase			:				Tuple 						=						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
            _lowercase			:				List[Any] 						=						os.path.join(lowerCamelCase_				,					'best-checkpoint'					)
            _lowercase			:				Tuple 						=						os.path.join(lowerCamelCase_				,					'stage-2'					)
            # Update arguments_dict
            _lowercase			:				Tuple 						=						model_path
            _lowercase			:				List[Any] 						=						data_files['train']
            _lowercase			:				List[Any] 						=						current_output_dir

            _lowercase			:				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_					)

        _lowercase			:				Tuple 						=						iteration
        _lowercase			:				Union[str, Any] 						=						data_dir_format(iteration + 1					)

        _lowercase			:				Optional[int] 						=						AutoConfig.from_pretrained(os.path.join(lowerCamelCase_				,					'best-checkpoint'					)					)
        _lowercase			:				Tuple 						=						config.idalabel
        _lowercase			:				int 						=						os.path.join(lowerCamelCase_				,					'eval_results_best-checkpoint.json'					)
        _lowercase			:				Dict 						=						os.path.join(lowerCamelCase_				,					'test_results_best-checkpoint.json'					)
        assert os.path.exists(lowerCamelCase_					)

        with open(lowerCamelCase_				,					'r'					) as f:
            _lowercase			:				List[Any] 						=						float(json.load(lowerCamelCase_					)[args.eval_metric]					)
        _lowercase			:				Any 						=						os.path.join(lowerCamelCase_				,					'infer_output_best-checkpoint.csv'					)
        assert os.path.exists(lowerCamelCase_					)
        # Loading the dataset from local csv or json files.
        _lowercase			:				Optional[Any] 						=						load_dataset(args.data_file_extension				,					data_files={'data': data_files['infer']}					)['data']
        _lowercase			:				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()

        _lowercase			:				str 						=						os.path.join(lowerCamelCase_				,					F'''train_pseudo.{args.data_file_extension}'''					)

        if args.evaluation_strategy != IntervalStrategy.NO.value:
            _lowercase			:				Any 						=						eval_result

            if best_iteration is None:
                _lowercase			:				Any 						=						new_iteration
                _lowercase			:				List[str] 						=						new_eval_result
            else:
                if new_eval_result - best_eval_result > args.early_stopping_threshold:
                    _lowercase			:				str 						=						new_iteration
                    _lowercase			:				List[Any] 						=						new_eval_result
                    _lowercase			:				int 						=						0
                else:
                    if new_eval_result == best_eval_result:
                        _lowercase			:				Optional[Any] 						=						new_iteration
                        _lowercase			:				List[Any] 						=						new_eval_result
                    early_stopping_patience_counter += 1

                if early_stopping_patience_counter >= args.early_stopping_patience:
                    _lowercase			:				int 						=						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'					)				,					)






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 _lowerCamelCase(		_a,					unittest.TestCase  ):
       lowercase_					:      Optional[int]       					=     """hf-internal-testing/tiny-random-OnnxStableDiffusionPipeline"""
       def      UpperCamelCase	(   self, lowerCamelCase=0)			->  str:




           """simple docstring"""

           _lowercase			:				Optional[int] 						=						np.random.RandomState(lowerCamelCase)
           _lowercase			:				Union[str, Any] 						=						{
               '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      UpperCamelCase	(   self)			->  List[str]:




           """simple docstring"""

           _lowercase			:				int 						=						OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint, provider='CPUExecutionProvider')
           pipe.set_progress_bar_config(disable=lowerCamelCase)

           _lowercase			:				List[str] 						=						self.get_dummy_inputs()
           _lowercase			:				Tuple 						=						pipe(**lowerCamelCase).images
           _lowercase			:				Optional[Any] 						=						image[0, -3:, -3:, -1]

           assert image.shape == (1, 1_28, 1_28, 3)
           _lowercase			:				List[Any] 						=						np.array([0.6_5_0_7_2, 0.5_8_4_9_2, 0.4_8_2_1_9, 0.5_5_5_2_1, 0.5_3_1_8_0, 0.5_5_9_3_9, 0.5_0_6_9_7, 0.3_9_8_0_0, 0.4_6_4_5_5])

           assert np.abs(image_slice.flatten() - expected_slice).max() < 1E-2
       def      UpperCamelCase	(   self)			->  List[Any]:




           """simple docstring"""

           _lowercase			:				str 						=						OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint, provider='CPUExecutionProvider')
           _lowercase			:				List[Any] 						=						PNDMScheduler.from_config(pipe.scheduler.config, skip_prk_steps=lowerCamelCase)
           pipe.set_progress_bar_config(disable=lowerCamelCase)

           _lowercase			:				Dict 						=						self.get_dummy_inputs()
           _lowercase			:				Optional[int] 						=						pipe(**lowerCamelCase).images
           _lowercase			:				Dict 						=						image[0, -3:, -3:, -1]

           assert image.shape == (1, 1_28, 1_28, 3)
           _lowercase			:				List[Any] 						=						np.array([0.6_5_8_6_3, 0.5_9_4_2_5, 0.4_9_3_2_6, 0.5_6_3_1_3, 0.5_3_8_7_5, 0.5_6_6_2_7, 0.5_1_0_6_5, 0.3_9_7_7_7, 0.4_6_3_3_0])

           assert np.abs(image_slice.flatten() - expected_slice).max() < 1E-2
       def      UpperCamelCase	(   self)			->  str:




           """simple docstring"""

           _lowercase			:				Optional[int] 						=						OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint, provider='CPUExecutionProvider')
           _lowercase			:				Optional[int] 						=						LMSDiscreteScheduler.from_config(pipe.scheduler.config)
           pipe.set_progress_bar_config(disable=lowerCamelCase)

           _lowercase			:				Dict 						=						self.get_dummy_inputs()
           _lowercase			:				Union[str, Any] 						=						pipe(**lowerCamelCase).images
           _lowercase			:				str 						=						image[0, -3:, -3:, -1]

           assert image.shape == (1, 1_28, 1_28, 3)
           _lowercase			:				Union[str, Any] 						=						np.array([0.5_3_7_5_5, 0.6_0_7_8_6, 0.4_7_4_0_2, 0.4_9_4_8_8, 0.5_1_8_6_9, 0.4_9_8_1_9, 0.4_7_9_8_5, 0.3_8_9_5_7, 0.4_4_2_7_9])

           assert np.abs(image_slice.flatten() - expected_slice).max() < 1E-2
       def      UpperCamelCase	(   self)			->  Optional[int]:




           """simple docstring"""

           _lowercase			:				Union[str, Any] 						=						OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint, provider='CPUExecutionProvider')
           _lowercase			:				Any 						=						EulerDiscreteScheduler.from_config(pipe.scheduler.config)
           pipe.set_progress_bar_config(disable=lowerCamelCase)

           _lowercase			:				List[Any] 						=						self.get_dummy_inputs()
           _lowercase			:				Any 						=						pipe(**lowerCamelCase).images
           _lowercase			:				Tuple 						=						image[0, -3:, -3:, -1]

           assert image.shape == (1, 1_28, 1_28, 3)
           _lowercase			:				Union[str, Any] 						=						np.array([0.5_3_7_5_5, 0.6_0_7_8_6, 0.4_7_4_0_2, 0.4_9_4_8_8, 0.5_1_8_6_9, 0.4_9_8_1_9, 0.4_7_9_8_5, 0.3_8_9_5_7, 0.4_4_2_7_9])

           assert np.abs(image_slice.flatten() - expected_slice).max() < 1E-2
       def      UpperCamelCase	(   self)			->  Dict:




           """simple docstring"""

           _lowercase			:				List[str] 						=						OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint, provider='CPUExecutionProvider')
           _lowercase			:				Optional[int] 						=						EulerAncestralDiscreteScheduler.from_config(pipe.scheduler.config)
           pipe.set_progress_bar_config(disable=lowerCamelCase)

           _lowercase			:				List[Any] 						=						self.get_dummy_inputs()
           _lowercase			:				Optional[int] 						=						pipe(**lowerCamelCase).images
           _lowercase			:				str 						=						image[0, -3:, -3:, -1]

           assert image.shape == (1, 1_28, 1_28, 3)
           _lowercase			:				List[str] 						=						np.array([0.5_3_8_1_7, 0.6_0_8_1_2, 0.4_7_3_8_4, 0.4_9_5_3_0, 0.5_1_8_9_4, 0.4_9_8_1_4, 0.4_7_9_8_4, 0.3_8_9_5_8, 0.4_4_2_7_1])

           assert np.abs(image_slice.flatten() - expected_slice).max() < 1E-2
       def      UpperCamelCase	(   self)			->  List[Any]:




           """simple docstring"""

           _lowercase			:				List[Any] 						=						OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint, provider='CPUExecutionProvider')
           _lowercase			:				Any 						=						DPMSolverMultistepScheduler.from_config(pipe.scheduler.config)
           pipe.set_progress_bar_config(disable=lowerCamelCase)

           _lowercase			:				Union[str, Any] 						=						self.get_dummy_inputs()
           _lowercase			:				Any 						=						pipe(**lowerCamelCase).images
           _lowercase			:				List[str] 						=						image[0, -3:, -3:, -1]

           assert image.shape == (1, 1_28, 1_28, 3)
           _lowercase			:				Any 						=						np.array([0.5_3_8_9_5, 0.6_0_8_0_8, 0.4_7_9_3_3, 0.4_9_6_0_8, 0.5_1_8_8_6, 0.4_9_9_5_0, 0.4_8_0_5_3, 0.3_8_9_5_7, 0.4_4_2_0_0])

           assert np.abs(image_slice.flatten() - expected_slice).max() < 1E-2
       def      UpperCamelCase	(   self)			->  int:




           """simple docstring"""

           _lowercase			:				Dict 						=						OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint, provider='CPUExecutionProvider')
           pipe.set_progress_bar_config(disable=lowerCamelCase)

           _lowercase			:				str 						=						self.get_dummy_inputs()
           _lowercase			:				Any 						=						3 * [inputs['prompt']]

           # forward
           _lowercase			:				int 						=						pipe(**lowerCamelCase)
           _lowercase			:				Optional[int] 						=						output.images[0, -3:, -3:, -1]

           _lowercase			:				int 						=						self.get_dummy_inputs()
           _lowercase			:				Union[str, Any] 						=						3 * [inputs.pop('prompt')]

           _lowercase			:				Union[str, Any] 						=						pipe.tokenizer(
               lowerCamelCase, padding='max_length', max_length=pipe.tokenizer.model_max_length, truncation=lowerCamelCase, return_tensors='np', )
           _lowercase			:				Tuple 						=						text_inputs['input_ids']

           _lowercase			:				Any 						=						pipe.text_encoder(input_ids=text_inputs.astype(np.intaa))[0]

           _lowercase			:				List[Any] 						=						prompt_embeds

           # forward
           _lowercase			:				Union[str, Any] 						=						pipe(**lowerCamelCase)
           _lowercase			:				Union[str, Any] 						=						output.images[0, -3:, -3:, -1]

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




       def      UpperCamelCase	(   self)			->  Optional[int]:




           """simple docstring"""

           _lowercase			:				Optional[Any] 						=						OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint, provider='CPUExecutionProvider')
           pipe.set_progress_bar_config(disable=lowerCamelCase)

           _lowercase			:				Optional[Any] 						=						self.get_dummy_inputs()
           _lowercase			:				Any 						=						3 * ['this is a negative prompt']
           _lowercase			:				str 						=						negative_prompt
           _lowercase			:				Optional[int] 						=						3 * [inputs['prompt']]

           # forward
           _lowercase			:				int 						=						pipe(**lowerCamelCase)
           _lowercase			:				str 						=						output.images[0, -3:, -3:, -1]

           _lowercase			:				Union[str, Any] 						=						self.get_dummy_inputs()
           _lowercase			:				str 						=						3 * [inputs.pop('prompt')]

           _lowercase			:				Optional[int] 						=						[]
           for p in [prompt, negative_prompt]:
               _lowercase			:				Tuple 						=						pipe.tokenizer(
                   lowerCamelCase, padding='max_length', max_length=pipe.tokenizer.model_max_length, truncation=lowerCamelCase, return_tensors='np', )
               _lowercase			:				Dict 						=						text_inputs['input_ids']

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

           _lowercase						,							_lowercase			:				str 						=						embeds

           # forward
           _lowercase			:				Dict 						=						pipe(**lowerCamelCase)
           _lowercase			:				Tuple 						=						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 _lowerCamelCase(		unittest.TestCase  ):
       @property
       def      UpperCamelCase	(   self)			->  Union[str, Any]:




           """simple docstring"""

           return (
               "CUDAExecutionProvider",
               {
                   "gpu_mem_limit": "15000000000",  # 15GB
                   "arena_extend_strategy": "kSameAsRequested",
               },
           )
       @property
       def      UpperCamelCase	(   self)			->  str:




           """simple docstring"""

           _lowercase			:				int 						=						ort.SessionOptions()
           _lowercase			:				str 						=						False
           return options
       def      UpperCamelCase	(   self)			->  Optional[Any]:




           """simple docstring"""

           _lowercase			:				Optional[int] 						=						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			:				List[Any] 						=						'A painting of a squirrel eating a burger'
           np.random.seed(0)
           _lowercase			:				Union[str, Any] 						=						sd_pipe([prompt], guidance_scale=6.0, num_inference_steps=10, output_type='np')
           _lowercase			:				Optional[Any] 						=						output.images

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

           assert image.shape == (1, 5_12, 5_12, 3)
           _lowercase			:				Union[str, Any] 						=						np.array([0.0_4_5_2, 0.0_3_9_0, 0.0_0_8_7, 0.0_3_5_0, 0.0_6_1_7, 0.0_3_6_4, 0.0_5_4_4, 0.0_5_2_3, 0.0_7_2_0])

           assert np.abs(image_slice.flatten() - expected_slice).max() < 1E-3
       def      UpperCamelCase	(   self)			->  Optional[int]:




           """simple docstring"""

           _lowercase			:				str 						=						DDIMScheduler.from_pretrained(
               'runwayml/stable-diffusion-v1-5', subfolder='scheduler', revision='onnx')
           _lowercase			:				str 						=						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			:				List[Any] 						=						'open neural network exchange'
           _lowercase			:				List[Any] 						=						np.random.RandomState(0)
           _lowercase			:				Optional[Any] 						=						sd_pipe([prompt], guidance_scale=7.5, num_inference_steps=10, generator=lowerCamelCase, output_type='np')
           _lowercase			:				Optional[Any] 						=						output.images
           _lowercase			:				List[Any] 						=						image[0, -3:, -3:, -1]

           assert image.shape == (1, 5_12, 5_12, 3)
           _lowercase			:				Optional[int] 						=						np.array([0.2_8_6_7, 0.1_9_7_4, 0.1_4_8_1, 0.7_2_9_4, 0.7_2_5_1, 0.6_6_6_7, 0.4_1_9_4, 0.5_6_4_2, 0.6_4_8_6])

           assert np.abs(image_slice.flatten() - expected_slice).max() < 1E-3
       def      UpperCamelCase	(   self)			->  List[str]:




           """simple docstring"""

           _lowercase			:				Dict 						=						LMSDiscreteScheduler.from_pretrained(
               'runwayml/stable-diffusion-v1-5', subfolder='scheduler', revision='onnx')
           _lowercase			:				Dict 						=						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			:				Tuple 						=						'open neural network exchange'
           _lowercase			:				str 						=						np.random.RandomState(0)
           _lowercase			:				Dict 						=						sd_pipe([prompt], guidance_scale=7.5, num_inference_steps=10, generator=lowerCamelCase, output_type='np')
           _lowercase			:				Optional[Any] 						=						output.images
           _lowercase			:				int 						=						image[0, -3:, -3:, -1]

           assert image.shape == (1, 5_12, 5_12, 3)
           _lowercase			:				Optional[Any] 						=						np.array([0.2_3_0_6, 0.1_9_5_9, 0.1_5_9_3, 0.6_5_4_9, 0.6_3_9_4, 0.5_4_0_8, 0.5_0_6_5, 0.6_0_1_0, 0.6_1_6_1])

           assert np.abs(image_slice.flatten() - expected_slice).max() < 1E-3
       def      UpperCamelCase	(   self)			->  int:




           """simple docstring"""

           _lowercase			:				List[Any] 						=						0

           def test_callback_fn(lowerCamelCase, lowerCamelCase, lowerCamelCase) -> None:
               _lowercase			:				List[str] 						=						True
               nonlocal number_of_steps
               number_of_steps += 1
               if step == 0:
                   assert latents.shape == (1, 4, 64, 64)
                   _lowercase			:				Any 						=						latents[0, -3:, -3:, -1]
                   _lowercase			:				Tuple 						=						np.array(
                       [-0.6_7_7_2, -0.3_8_3_5, -1.2_4_5_6, 0.1_9_0_5, -1.0_9_7_4, 0.6_9_6_7, -1.9_3_5_3, 0.0_1_7_8, 1.0_1_6_7])

                   assert np.abs(latents_slice.flatten() - expected_slice).max() < 1E-3
               elif step == 5:
                   assert latents.shape == (1, 4, 64, 64)
                   _lowercase			:				List[Any] 						=						latents[0, -3:, -3:, -1]
                   _lowercase			:				str 						=						np.array(
                       [-0.3_3_5_1, 0.2_2_4_1, -0.1_8_3_7, -0.2_3_2_5, -0.6_5_7_7, 0.3_3_9_3, -0.0_2_4_1, 0.5_8_9_9, 1.3_8_7_5])

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

           _lowercase			:				Any 						=						False

           _lowercase			:				int 						=						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			:				Any 						=						'Andromeda galaxy in a bottle'

           _lowercase			:				str 						=						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      UpperCamelCase	(   self)			->  Tuple:




           """simple docstring"""

           _lowercase			:				List[str] 						=						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			:				Optional[int] 						=						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			:				Any 						=						OnnxStableDiffusionPipeline.from_pretrained(lowerCamelCase)

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


