




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  A		(  __UpperCamelCase      ,				unittest.TestCase     ):
		__snake_case									=   'hf-internal-testing/tiny-random-OnnxStableDiffusionPipeline'



		def        SCREAMING_SNAKE_CASE__			(     self,	UpperCamelCase__=0   ):







								"""simple docstring"""



								lowerCAmelCase_      			=  np.random.RandomState(lowerCAmelCase__   )
								lowerCAmelCase_      			=  {
								    '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        SCREAMING_SNAKE_CASE__			(     self   ):







								"""simple docstring"""



								lowerCAmelCase_      			=  OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint,	provider='''CPUExecutionProvider'''   )
								pipe.set_progress_bar_config(disable=lowerCAmelCase__   )

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

								assert image.shape == (1, 128, 128, 3)
								lowerCAmelCase_      			=  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        SCREAMING_SNAKE_CASE__			(     self   ):







								"""simple docstring"""



								lowerCAmelCase_      			=  OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint,	provider='''CPUExecutionProvider'''   )
								lowerCAmelCase_      			=  PNDMScheduler.from_config(pipe.scheduler.config,	skip_prk_steps=lowerCAmelCase__   )
								pipe.set_progress_bar_config(disable=lowerCAmelCase__   )

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

								assert image.shape == (1, 128, 128, 3)
								lowerCAmelCase_      			=  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        SCREAMING_SNAKE_CASE__			(     self   ):







								"""simple docstring"""



								lowerCAmelCase_      			=  OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint,	provider='''CPUExecutionProvider'''   )
								lowerCAmelCase_      			=  LMSDiscreteScheduler.from_config(pipe.scheduler.config   )
								pipe.set_progress_bar_config(disable=lowerCAmelCase__   )

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

								assert image.shape == (1, 128, 128, 3)
								lowerCAmelCase_      			=  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        SCREAMING_SNAKE_CASE__			(     self   ):







								"""simple docstring"""



								lowerCAmelCase_      			=  OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint,	provider='''CPUExecutionProvider'''   )
								lowerCAmelCase_      			=  EulerDiscreteScheduler.from_config(pipe.scheduler.config   )
								pipe.set_progress_bar_config(disable=lowerCAmelCase__   )

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

								assert image.shape == (1, 128, 128, 3)
								lowerCAmelCase_      			=  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        SCREAMING_SNAKE_CASE__			(     self   ):







								"""simple docstring"""



								lowerCAmelCase_      			=  OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint,	provider='''CPUExecutionProvider'''   )
								lowerCAmelCase_      			=  EulerAncestralDiscreteScheduler.from_config(pipe.scheduler.config   )
								pipe.set_progress_bar_config(disable=lowerCAmelCase__   )

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

								assert image.shape == (1, 128, 128, 3)
								lowerCAmelCase_      			=  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        SCREAMING_SNAKE_CASE__			(     self   ):







								"""simple docstring"""



								lowerCAmelCase_      			=  OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint,	provider='''CPUExecutionProvider'''   )
								lowerCAmelCase_      			=  DPMSolverMultistepScheduler.from_config(pipe.scheduler.config   )
								pipe.set_progress_bar_config(disable=lowerCAmelCase__   )

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

								assert image.shape == (1, 128, 128, 3)
								lowerCAmelCase_      			=  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        SCREAMING_SNAKE_CASE__			(     self   ):







								"""simple docstring"""



								lowerCAmelCase_      			=  OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint,	provider='''CPUExecutionProvider'''   )
								pipe.set_progress_bar_config(disable=lowerCAmelCase__   )

								lowerCAmelCase_      			=  self.get_dummy_inputs()
								lowerCAmelCase_      			=  3 * [inputs['prompt']]

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

								lowerCAmelCase_      			=  self.get_dummy_inputs()
								lowerCAmelCase_      			=  3 * [inputs.pop('''prompt'''   )]

								lowerCAmelCase_      			=  pipe.tokenizer(
								    lowerCAmelCase__,	padding='''max_length''',	max_length=pipe.tokenizer.model_max_length,	truncation=lowerCAmelCase__,	return_tensors='''np''',	)
								lowerCAmelCase_      			=  text_inputs['input_ids']

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

								lowerCAmelCase_      			=  prompt_embeds

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

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







		def        SCREAMING_SNAKE_CASE__			(     self   ):







								"""simple docstring"""



								lowerCAmelCase_      			=  OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint,	provider='''CPUExecutionProvider'''   )
								pipe.set_progress_bar_config(disable=lowerCAmelCase__   )

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

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

								lowerCAmelCase_      			=  self.get_dummy_inputs()
								lowerCAmelCase_      			=  3 * [inputs.pop('''prompt'''   )]

								lowerCAmelCase_      			=  []
								for p in [prompt, negative_prompt]:
														lowerCAmelCase_      			=  pipe.tokenizer(
														    lowerCAmelCase__,	padding='''max_length''',	max_length=pipe.tokenizer.model_max_length,	truncation=lowerCAmelCase__,	return_tensors='''np''',	)
														lowerCAmelCase_      			=  text_inputs['input_ids']

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

								lowerCAmelCase_      			=  embeds

								# forward
								lowerCAmelCase_      			=  pipe(**lowerCAmelCase__   )
								lowerCAmelCase_      			=  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  A		(  unittest.TestCase     ):



		@property
		def        SCREAMING_SNAKE_CASE__			(     self   ):







								"""simple docstring"""



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



		@property
		def        SCREAMING_SNAKE_CASE__			(     self   ):







								"""simple docstring"""



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



		def        SCREAMING_SNAKE_CASE__			(     self   ):







								"""simple docstring"""



								lowerCAmelCase_      			=  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__   )

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

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

								assert image.shape == (1, 512, 512, 3)
								lowerCAmelCase_      			=  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        SCREAMING_SNAKE_CASE__			(     self   ):







								"""simple docstring"""



								lowerCAmelCase_      			=  DDIMScheduler.from_pretrained(
								    '''runwayml/stable-diffusion-v1-5''',	subfolder='''scheduler''',	revision='''onnx'''   )
								lowerCAmelCase_      			=  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__   )

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

								assert image.shape == (1, 512, 512, 3)
								lowerCAmelCase_      			=  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        SCREAMING_SNAKE_CASE__			(     self   ):







								"""simple docstring"""



								lowerCAmelCase_      			=  LMSDiscreteScheduler.from_pretrained(
								    '''runwayml/stable-diffusion-v1-5''',	subfolder='''scheduler''',	revision='''onnx'''   )
								lowerCAmelCase_      			=  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__   )

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

								assert image.shape == (1, 512, 512, 3)
								lowerCAmelCase_      			=  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        SCREAMING_SNAKE_CASE__			(     self   ):







								"""simple docstring"""



								lowerCAmelCase_      			=  0

								def test_callback_fn(UpperCamelCase__,	UpperCamelCase__,	UpperCamelCase__   ) -> None:
														lowerCAmelCase_      			=  True
														nonlocal number_of_steps
														number_of_steps += 1
														if step == 0:
																				assert latents.shape == (1, 4, 64, 64)
																				lowerCAmelCase_      			=  latents[0, -3:, -3:, -1]
																				lowerCAmelCase_      			=  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)
																				lowerCAmelCase_      			=  latents[0, -3:, -3:, -1]
																				lowerCAmelCase_      			=  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

								lowerCAmelCase_      			=  False

								lowerCAmelCase_      			=  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__   )

								lowerCAmelCase_      			=  'Andromeda galaxy in a bottle'

								lowerCAmelCase_      			=  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        SCREAMING_SNAKE_CASE__			(     self   ):







								"""simple docstring"""



								lowerCAmelCase_      			=  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

								lowerCAmelCase_      			=  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__   )
														lowerCAmelCase_      			=  OnnxStableDiffusionPipeline.from_pretrained(lowerCAmelCase__   )

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








import argparse
import json
import os
import time
import zipfile

from get_ci_error_statistics import download_artifact, get_artifacts_links

from transformers import logging


__lowerCAmelCase		       =							logging.get_logger(__name__)




def    snake_case_     (  snake_case			,							snake_case     )  ->				List[str]:
       lowercase__:   List[str]  			=   set()
       lowercase__:   List[Any]  			=   []

       def parse_line(snake_case     ):
              for line in fp:
                     if isinstance(snake_case			,							snake_case     ):
                            lowercase__:   Optional[Any]  			=   line.decode('UTF-8'     )
                     if "warnings summary (final)" in line:
                            continue
                     # This means we are outside the body of a warning
                     elif not line.startswith(' '     ):
                            # process a single warning and move it to `selected_warnings`.
                            if len(snake_case     ) > 0:
                                   lowercase__:   List[str]  			=   '\n'.join(snake_case     )
                                   # Only keep the warnings specified in `targets`
                                   if any(f': {x}: ' in warning for x in targets     ):
                                          selected_warnings.add(snake_case     )
                                   buffer.clear()
                            continue
                     else:
                            lowercase__:   Union[str, Any]  			=   line.strip()
                            buffer.append(snake_case     )

       if from_gh:
              for filename in os.listdir(snake_case     ):
                     lowercase__:   Dict  			=   os.path.join(snake_case			,							snake_case     )
                     if not os.path.isdir(snake_case     ):
                            # read the file
                            if filename != "warnings.txt":
                                   continue
                            with open(snake_case     ) as fp:
                                   parse_line(snake_case     )
       else:
              try:
                     with zipfile.ZipFile(snake_case     ) as z:
                            for filename in z.namelist():
                                   if not os.path.isdir(snake_case     ):
                                          # read the file
                                          if filename != "warnings.txt":
                                                 continue
                                          with z.open(snake_case     ) as fp:
                                                 parse_line(snake_case     )
              except Exception:
                     logger.warning(
                         f'{artifact_path} is either an invalid zip file or something else wrong. This file is skipped.'     )

       return selected_warnings




def    snake_case_     (  snake_case			,							snake_case     )  ->				Any:
       lowercase__:   Optional[Any]  			=   set()

       lowercase__:   int  			=   [os.path.join(snake_case			,							snake_case     ) for p in os.listdir(snake_case     ) if (p.endswith('.zip'     ) or from_gh)]
       for p in paths:
              selected_warnings.update(extract_warnings_from_single_artifact(snake_case			,							snake_case     )     )

       return selected_warnings


if __name__ == "__main__":




 def    snake_case_     (  snake_case     )  ->				str:
        return values.split(','     )

 __lowerCAmelCase		       =							argparse.ArgumentParser()
 # Required parameters
 parser.add_argument('''--workflow_run_id''', type=str, required=True, help='''A GitHub Actions workflow run id.''')
 parser.add_argument(
     '''--output_dir''',
     type=str,
     required=True,
     help='''Where to store the downloaded artifacts and other result files.''',
 )
 parser.add_argument('''--token''', default=None, type=str, help='''A token that has actions:read permission.''')
 # optional parameters
 parser.add_argument(
     '''--targets''',
     default='''DeprecationWarning,UserWarning,FutureWarning''',
     type=list_str,
     help='''Comma-separated list of target warning(s) which we want to extract.''',
 )
 parser.add_argument(
     '''--from_gh''',
     action='''store_true''',
     help='''If running from a GitHub action workflow and collecting warnings from its artifacts.''',
 )

 __lowerCAmelCase		       =							parser.parse_args()

 __lowerCAmelCase		       =							args.from_gh
 if from_gh:
  # The artifacts have to be downloaded using `actions/download-artifact@v3`
  pass
 else:
  os.makedirs(args.output_dir, exist_ok=True)

  # get download links
  __lowerCAmelCase		       =							get_artifacts_links(args.workflow_run_id, token=args.token)
  with open(os.path.join(args.output_dir, '''artifacts.json'''), '''w''', encoding='''UTF-8''') as fp:
   json.dump(artifacts, fp, ensure_ascii=False, indent=4)

  # download artifacts
  for idx, (name, url) in enumerate(artifacts.items()):
   print(name)
   print(url)
   print('''=''' * 80)
   download_artifact(name, url, args.output_dir, args.token)
   # Be gentle to GitHub
   time.sleep(1)

    # extract warnings from artifacts
 __lowerCAmelCase		       =							extract_warnings(args.output_dir, args.targets)
 __lowerCAmelCase		       =							sorted(selected_warnings)
 with open(os.path.join(args.output_dir, '''selected_warnings.json'''), '''w''', encoding='''UTF-8''') as fp:
  json.dump(selected_warnings, fp, ensure_ascii=False, indent=4)


