




'''simple docstring'''


import random
import unittest

import numpy as np
import torch

from diffusers import (
    DPMSolverMultistepScheduler,
    EulerAncestralDiscreteScheduler,
    EulerDiscreteScheduler,
    LMSDiscreteScheduler,
    OnnxStableDiffusionUpscalePipeline,
    PNDMScheduler,
)
from diffusers.utils import floats_tensor
from diffusers.utils.testing_utils import (
    is_onnx_available,
    load_image,
    nightly,
    require_onnxruntime,
    require_torch_gpu,
)

from ..test_pipelines_onnx_common import OnnxPipelineTesterMixin


if is_onnx_available():
     import onnxruntime as ort






class 		lowerCAmelCase_     (    UpperCamelCase_    , unittest.TestCase				):







  '''simple docstring'''

  UpperCamelCase_       :	int				       =							"""ssube/stable-diffusion-x4-upscaler-onnx"""



  def  _snake_case			(      self							:				Tuple ,   SCREAMING_SNAKE_CASE_							:				Optional[Any]=0     )       ->						Union[str, Any]:







       '''simple docstring'''



       A:						Optional[Any]						=						floats_tensor((1, 3, 1_28, 1_28) ,   rng=random.Random(A_     )     )
       A:						Dict						=						torch.manual_seed(A_     )
       A:						Dict						=						{
           '''prompt''': '''A painting of a squirrel eating a burger''',
           '''image''': image,
           '''generator''': generator,
           '''num_inference_steps''': 3,
           '''guidance_scale''': 7.5,
           '''output_type''': '''numpy''',
       }
       return inputs



  def  _snake_case			(      self							:				List[str]     )       ->						Union[str, Any]:







       '''simple docstring'''



       A:						int						=						OnnxStableDiffusionUpscalePipeline.from_pretrained(self.hub_checkpoint ,   provider='''CPUExecutionProvider'''     )
       pipe.set_progress_bar_config(disable=A_     )

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

       # started as 128, should now be 512
       assert image.shape == (1, 5_12, 5_12, 3)
       A:						List[str]						=						np.array(
           [0.697_4782, 0.6890_2093, 0.7013_5885, 0.758_3618, 0.780_4545, 0.785_4912, 0.7866_7426, 0.7874_3863, 0.7807_0223]     )
       assert np.abs(image_slice - expected_slice     ).max() < 1E-1



  def  _snake_case			(      self							:				Union[str, Any]     )       ->						Optional[int]:







       '''simple docstring'''



       A:						Union[str, Any]						=						OnnxStableDiffusionUpscalePipeline.from_pretrained(self.hub_checkpoint ,   provider='''CPUExecutionProvider'''     )
       A:						List[str]						=						PNDMScheduler.from_config(pipe.scheduler.config ,   skip_prk_steps=A_     )
       pipe.set_progress_bar_config(disable=A_     )

       A:						Any						=						self.get_dummy_inputs()
       A:						int						=						pipe(**A_     ).images
       A:						List[Any]						=						image[0, -3:, -3:, -1]

       assert image.shape == (1, 5_12, 5_12, 3)
       A:						str						=						np.array(
           [0.689_8892, 0.5924_0556, 0.5249_9527, 0.5886_6215, 0.5225_8235, 0.5257_2715, 0.6241_4473, 0.617_4387, 0.621_4964]     )
       assert np.abs(image_slice.flatten() - expected_slice     ).max() < 1E-1



  def  _snake_case			(      self							:				Dict     )       ->						List[Any]:







       '''simple docstring'''



       A:						List[Any]						=						OnnxStableDiffusionUpscalePipeline.from_pretrained(self.hub_checkpoint ,   provider='''CPUExecutionProvider'''     )
       A:						int						=						DPMSolverMultistepScheduler.from_config(pipe.scheduler.config     )
       pipe.set_progress_bar_config(disable=A_     )

       A:						Optional[Any]						=						self.get_dummy_inputs()
       A:						Optional[int]						=						pipe(**A_     ).images
       A:						Union[str, Any]						=						image[0, -3:, -3:, -1]

       assert image.shape == (1, 5_12, 5_12, 3)
       A:						Optional[int]						=						np.array(
           [0.765_9278, 0.7643_7664, 0.7557_9107, 0.769_1116, 0.7766_6986, 0.772_7672, 0.775_8664, 0.781_2226, 0.7694_2515]     )

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



  def  _snake_case			(      self							:				Optional[Any]     )       ->						Union[str, Any]:







       '''simple docstring'''



       A:						Union[str, Any]						=						OnnxStableDiffusionUpscalePipeline.from_pretrained(self.hub_checkpoint ,   provider='''CPUExecutionProvider'''     )
       A:						Tuple						=						EulerDiscreteScheduler.from_config(pipe.scheduler.config     )
       pipe.set_progress_bar_config(disable=A_     )

       A:						str						=						self.get_dummy_inputs()
       A:						int						=						pipe(**A_     ).images
       A:						Union[str, Any]						=						image[0, -3:, -3:, -1]

       assert image.shape == (1, 5_12, 5_12, 3)
       A:						Tuple						=						np.array(
           [0.697_4782, 0.6890_2093, 0.7013_5885, 0.758_3618, 0.780_4545, 0.785_4912, 0.7866_7426, 0.7874_3863, 0.7807_0223]     )
       assert np.abs(image_slice.flatten() - expected_slice     ).max() < 1E-1




  def  _snake_case			(      self							:				Optional[int]     )       ->						List[Any]:







       '''simple docstring'''



       A:						int						=						OnnxStableDiffusionUpscalePipeline.from_pretrained(self.hub_checkpoint ,   provider='''CPUExecutionProvider'''     )
       A:						List[Any]						=						EulerAncestralDiscreteScheduler.from_config(pipe.scheduler.config     )
       pipe.set_progress_bar_config(disable=A_     )

       A:						Tuple						=						self.get_dummy_inputs()
       A:						Dict						=						pipe(**A_     ).images
       A:						List[str]						=						image[0, -3:, -3:, -1]

       assert image.shape == (1, 5_12, 5_12, 3)
       A:						List[str]						=						np.array(
           [0.7742_4496, 0.77_3601, 0.764_5288, 0.776_9598, 0.777_2739, 0.773_8688, 0.7818_7233, 0.7787_9584, 0.76_7043]     )

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






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







  '''simple docstring'''

  @property
  def  _snake_case			(      self							:				Union[str, Any]     )       ->						Optional[int]:







       '''simple docstring'''



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



  @property
  def  _snake_case			(      self							:				Any     )       ->						Optional[Any]:







       '''simple docstring'''



       A:						Dict						=						ort.SessionOptions()
       A:						Optional[int]						=						False
       return options



  def  _snake_case			(      self							:				str     )       ->						List[str]:







       '''simple docstring'''



       A:						Any						=						load_image(
           '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
           '''/img2img/sketch-mountains-input.jpg'''     )
       A:						Tuple						=						init_image.resize((1_28, 1_28)     )
       # using the PNDM scheduler by default
       A:						Union[str, Any]						=						OnnxStableDiffusionUpscalePipeline.from_pretrained(
           '''ssube/stable-diffusion-x4-upscaler-onnx''' ,   provider=self.gpu_provider ,   sess_options=self.gpu_options ,   )
       pipe.set_progress_bar_config(disable=A_     )

       A:						Optional[Any]						=						'''A fantasy landscape, trending on artstation'''

       A:						str						=						torch.manual_seed(0     )
       A:						Optional[Any]						=						pipe(
           prompt=A_ ,   image=A_ ,   guidance_scale=7.5 ,   num_inference_steps=10 ,   generator=A_ ,   output_type='''np''' ,   )
       A:						Union[str, Any]						=						output.images
       A:						List[Any]						=						images[0, 2_55:2_58, 3_83:3_86, -1]

       assert images.shape == (1, 5_12, 5_12, 3)
       A:						Optional[Any]						=						np.array([0.4883, 0.4947, 0.4980, 0.4975, 0.4982, 0.4980, 0.5000, 0.5006, 0.4972]     )
       # TODO: lower the tolerance after finding the cause of onnxruntime reproducibility issues

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




  def  _snake_case			(      self							:				Dict     )       ->						Union[str, Any]:







       '''simple docstring'''



       A:						Tuple						=						load_image(
           '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
           '''/img2img/sketch-mountains-input.jpg'''     )
       A:						List[str]						=						init_image.resize((1_28, 1_28)     )
       A:						Dict						=						LMSDiscreteScheduler.from_pretrained(
           '''ssube/stable-diffusion-x4-upscaler-onnx''' ,   subfolder='''scheduler'''     )
       A:						List[str]						=						OnnxStableDiffusionUpscalePipeline.from_pretrained(
           '''ssube/stable-diffusion-x4-upscaler-onnx''' ,   scheduler=A_ ,   provider=self.gpu_provider ,   sess_options=self.gpu_options ,   )
       pipe.set_progress_bar_config(disable=A_     )

       A:						Optional[Any]						=						'''A fantasy landscape, trending on artstation'''

       A:						Any						=						torch.manual_seed(0     )
       A:						List[str]						=						pipe(
           prompt=A_ ,   image=A_ ,   guidance_scale=7.5 ,   num_inference_steps=20 ,   generator=A_ ,   output_type='''np''' ,   )
       A:						List[Any]						=						output.images
       A:						Tuple						=						images[0, 2_55:2_58, 3_83:3_86, -1]

       assert images.shape == (1, 5_12, 5_12, 3)
       A:						Optional[int]						=						np.array(
           [0.5017_3753, 0.5022_3356, 0.50_2039, 0.5023_3036, 0.502_3725, 0.502_2601, 0.501_8758, 0.5023_4085, 0.5024_1566]     )
       # TODO: lower the tolerance after finding the cause of onnxruntime reproducibility issues

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











from pathlib import Path

import fire


def 			UpperCamelCase(  __UpperCamelCase							:		str     ,__UpperCamelCase							:		str     ,__UpperCamelCase							:		int ):
		lowerCAmelCase_			:     List[str]									= Path(__UpperCamelCase )
		lowerCAmelCase_			:     Union[str, Any]									= Path(__UpperCamelCase )
		dest_dir.mkdir(exist_ok=__UpperCamelCase )
		for path in src_dir.iterdir():
				lowerCAmelCase_			:     Optional[Any]									= [x.rstrip() for x in list(path.open().readlines() )][:n]
				lowerCAmelCase_			:     List[str]									= dest_dir.joinpath(path.name )
				print(__UpperCamelCase )
				dest_path.open('''w''' ).write('''\n'''.join(__UpperCamelCase ) )


if __name__ == "__main__":
						fire.Fire(minify)





