


"""simple docstring"""


import json
import os
from pathlib import Path

import pytest

from datasets.download.download_config import DownloadConfig
from datasets.download.download_manager import DownloadManager
from datasets.utils.file_utils import hash_url_to_filename


A     :   Optional[Any]			    =			"http://www.mocksite.com/file1.txt"
A     :   Optional[Any]			    =			"\"text\": [\"foo\", \"foo\"]"
A     :   List[str]			    =			"6d8ce9aa78a471c7477201efbeabd3bb01ac2e7d100a6dc024ba1608361f90a8"
class   _UpperCamelCase  :
							'''simple docstring'''







							__UpperCAmelCase						:    Optional[Any]				      =2_0_0
							__UpperCAmelCase						:    List[str]				      ={'Content-Length': '100'}
							__UpperCAmelCase						:    int				      ={}

							def 			snake_case				(     self    ,   **__a	):
										return [bytes(a__    ,   "utf-8"	)]


def _lowerCamelCase	(		*_UpperCamelCase    , **_UpperCamelCase      ):

			'''simple docstring'''




			return MockResponse()


@pytest.mark.parametrize("urls_type"    , [str, list, dict]      )
def _lowerCamelCase	(		_UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase      ):

			'''simple docstring'''




			import requests

			monkeypatch.setattr(snake_case_    , "request"    , snake_case_      )

			__lowerCAmelCase				    =			URL
			if issubclass(snake_case_    , snake_case_      ):
						__lowerCAmelCase				    =			url
			elif issubclass(snake_case_    , snake_case_      ):
						__lowerCAmelCase				    =			[url]
			elif issubclass(snake_case_    , snake_case_      ):
						__lowerCAmelCase				    =			{"train": url}
			__lowerCAmelCase				    =			"dummy"
			__lowerCAmelCase				    =			"downloads"
			__lowerCAmelCase				    =			tmp_path
			__lowerCAmelCase				    =			DownloadConfig(
			    cache_dir=os.path.join(snake_case_    , snake_case_      )    , use_etag=snake_case_    , )
			__lowerCAmelCase				    =			DownloadManager(dataset_name=snake_case_    , download_config=snake_case_      )
			__lowerCAmelCase				    =			dl_manager.download(snake_case_      )
			__lowerCAmelCase				    =			urls
			for downloaded_paths in [downloaded_paths]:
						if isinstance(snake_case_    , snake_case_      ):
									__lowerCAmelCase				    =			[downloaded_paths]
									__lowerCAmelCase				    =			[urls]
						elif isinstance(snake_case_    , snake_case_      ):
									assert "train" in downloaded_paths.keys()
									__lowerCAmelCase				    =			downloaded_paths.values()
									__lowerCAmelCase				    =			urls.values()
						assert downloaded_paths
						for downloaded_path, input_url in zip(snake_case_    , snake_case_      ):
									assert downloaded_path == dl_manager.downloaded_paths[input_url]
									__lowerCAmelCase				    =			Path(snake_case_      )
									__lowerCAmelCase				    =			downloaded_path.parts
									assert parts[-1] == HASH
									assert parts[-2] == cache_subdir
									assert downloaded_path.exists()
									__lowerCAmelCase				    =			downloaded_path.read_text()
									assert content == CONTENT
									__lowerCAmelCase				    =			downloaded_path.with_suffix(".json"      )
									assert metadata_downloaded_path.exists()
									__lowerCAmelCase				    =			json.loads(metadata_downloaded_path.read_text()      )
									assert metadata_content == {"url": URL, "etag": None}


@pytest.mark.parametrize("paths_type"    , [str, list, dict]      )
def _lowerCamelCase	(		_UpperCamelCase    , _UpperCamelCase    , _UpperCamelCase      ):

			'''simple docstring'''




			__lowerCAmelCase				    =			str(snake_case_      )
			if issubclass(snake_case_    , snake_case_      ):
						__lowerCAmelCase				    =			filename
			elif issubclass(snake_case_    , snake_case_      ):
						__lowerCAmelCase				    =			[filename]
			elif issubclass(snake_case_    , snake_case_      ):
						__lowerCAmelCase				    =			{"train": filename}
			__lowerCAmelCase				    =			"dummy"
			__lowerCAmelCase				    =			xz_file.parent
			__lowerCAmelCase				    =			"extracted"
			__lowerCAmelCase				    =			DownloadConfig(
			    cache_dir=snake_case_    , use_etag=snake_case_    , )
			__lowerCAmelCase				    =			DownloadManager(dataset_name=snake_case_    , download_config=snake_case_      )
			__lowerCAmelCase				    =			dl_manager.extract(snake_case_      )
			__lowerCAmelCase				    =			paths
			for extracted_paths in [extracted_paths]:
						if isinstance(snake_case_    , snake_case_      ):
									__lowerCAmelCase				    =			[extracted_paths]
									__lowerCAmelCase				    =			[paths]
						elif isinstance(snake_case_    , snake_case_      ):
									assert "train" in extracted_paths.keys()
									__lowerCAmelCase				    =			extracted_paths.values()
									__lowerCAmelCase				    =			paths.values()
						assert extracted_paths
						for extracted_path, input_path in zip(snake_case_    , snake_case_      ):
									assert extracted_path == dl_manager.extracted_paths[input_path]
									__lowerCAmelCase				    =			Path(snake_case_      )
									__lowerCAmelCase				    =			extracted_path.parts
									assert parts[-1] == hash_url_to_filename(snake_case_    , etag=snake_case_      )
									assert parts[-2] == extracted_subdir
									assert extracted_path.exists()
									__lowerCAmelCase				    =			extracted_path.read_text()
									__lowerCAmelCase				    =			text_file.read_text()
									assert extracted_file_content == expected_file_content


def _lowerCamelCase	(		_UpperCamelCase    , _UpperCamelCase      ):

			'''simple docstring'''




			assert path.endswith(".jsonl"      )
			for num_items, line in enumerate(snake_case_    , start=1      ):
						__lowerCAmelCase				    =			json.loads(line.decode("utf-8"      )      )
						assert item.keys() == {"col_1", "col_2", "col_3"}
			assert num_items == 4


@pytest.mark.parametrize("archive_jsonl"    , ["tar_jsonl_path", "zip_jsonl_path"]      )
def _lowerCamelCase	(		_UpperCamelCase    , _UpperCamelCase      ):

			'''simple docstring'''




			__lowerCAmelCase				    =			request.getfixturevalue(snake_case_      )
			__lowerCAmelCase				    =			DownloadManager()
			for num_jsonl, (path, file) in enumerate(dl_manager.iter_archive(snake_case_      )    , start=1      ):
						_test_jsonl(snake_case_    , snake_case_      )
			assert num_jsonl == 2


@pytest.mark.parametrize("archive_nested_jsonl"    , ["tar_nested_jsonl_path", "zip_nested_jsonl_path"]      )
def _lowerCamelCase	(		_UpperCamelCase    , _UpperCamelCase      ):

			'''simple docstring'''




			__lowerCAmelCase				    =			request.getfixturevalue(snake_case_      )
			__lowerCAmelCase				    =			DownloadManager()
			for num_tar, (path, file) in enumerate(dl_manager.iter_archive(snake_case_      )    , start=1      ):
						for num_jsonl, (subpath, subfile) in enumerate(dl_manager.iter_archive(snake_case_      )    , start=1      ):
									_test_jsonl(snake_case_    , snake_case_      )
			assert num_tar == 1
			assert num_jsonl == 2


def _lowerCamelCase	(		_UpperCamelCase      ):

			'''simple docstring'''




			__lowerCAmelCase				    =			DownloadManager()
			for num_file, file in enumerate(dl_manager.iter_files(snake_case_      )    , start=1      ):
						assert os.path.basename(snake_case_      ) == ("test.txt" if num_file == 1 else "train.txt")
			assert num_file == 2











import random
import unittest

import numpy as np

from diffusers import (
    DPMSolverMultistepScheduler,
    EulerAncestralDiscreteScheduler,
    EulerDiscreteScheduler,
    LMSDiscreteScheduler,
    OnnxStableDiffusionImgaImgPipeline,
    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 		SCREAMING_SNAKE_CASE__				(       _UpperCAmelCase						,      unittest.TestCase  ):
      A_						:				Tuple							=    'hf-internal-testing/tiny-random-OnnxStableDiffusionPipeline'


      def    a (self       :      int				,			a__       :      List[Any]=0      ):





         """simple docstring"""



         __snake_case							       =   floats_tensor((1, 3, 128, 128)				,			rng=random.Random(a__      )      )
         __snake_case							       =   np.random.RandomState(a__      )
         __snake_case							       =   {
             '''prompt''': '''A painting of a squirrel eating a burger''',
             '''image''': image,
             '''generator''': generator,
             '''num_inference_steps''': 3,
             '''strength''': 0.7_5,
             '''guidance_scale''': 7.5,
             '''output_type''': '''numpy''',
         }
         return inputs


      def    a (self       :      List[Any]      ):





         """simple docstring"""



         __snake_case							       =   OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint				,			provider='''CPUExecutionProvider'''      )
         pipe.set_progress_bar_config(disable=a__      )

         __snake_case							       =   self.get_dummy_inputs()
         __snake_case							       =   pipe(**a__      ).images
         __snake_case							       =   image[0, -3:, -3:, -1].flatten()

         assert image.shape == (1, 128, 128, 3)
         __snake_case							       =   np.array([0.6_9_6_4_3, 0.5_8_4_8_4, 0.5_0_3_1_4, 0.5_8_7_6_0, 0.5_5_3_6_8, 0.5_9_6_4_3, 0.5_1_5_2_9, 0.4_1_2_1_7, 0.4_9_0_8_7]      )
         assert np.abs(image_slice - expected_slice      ).max() < 1E-1


      def    a (self       :      Dict      ):





         """simple docstring"""



         __snake_case							       =   OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint				,			provider='''CPUExecutionProvider'''      )
         __snake_case							       =   PNDMScheduler.from_config(pipe.scheduler.config				,			skip_prk_steps=a__      )
         pipe.set_progress_bar_config(disable=a__      )

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

         assert image.shape == (1, 128, 128, 3)
         __snake_case							       =   np.array([0.6_1_7_3_7, 0.5_4_6_4_2, 0.5_3_1_8_3, 0.5_4_4_6_5, 0.5_2_7_4_2, 0.6_0_5_2_5, 0.4_9_9_6_9, 0.4_0_6_5_5, 0.4_8_1_5_4]      )

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


      def    a (self       :      List[str]      ):





         """simple docstring"""



         __snake_case							       =   OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint				,			provider='''CPUExecutionProvider'''      )
         __snake_case							       =   LMSDiscreteScheduler.from_config(pipe.scheduler.config      )
         pipe.set_progress_bar_config(disable=a__      )

         # warmup pass to apply optimizations
         __snake_case							       =   pipe(**self.get_dummy_inputs()      )

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

         assert image.shape == (1, 128, 128, 3)
         __snake_case							       =   np.array([0.5_2_7_6_1, 0.5_9_9_7_7, 0.4_9_0_3_3, 0.4_9_6_1_9, 0.5_4_2_8_2, 0.5_0_3_1_1, 0.4_7_6_0_0, 0.4_0_9_1_8, 0.4_5_2_0_3]      )

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


      def    a (self       :      Any      ):





         """simple docstring"""



         __snake_case							       =   OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint				,			provider='''CPUExecutionProvider'''      )
         __snake_case							       =   EulerDiscreteScheduler.from_config(pipe.scheduler.config      )
         pipe.set_progress_bar_config(disable=a__      )

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

         assert image.shape == (1, 128, 128, 3)
         __snake_case							       =   np.array([0.5_2_9_1_1, 0.6_0_0_0_4, 0.4_9_2_2_9, 0.4_9_8_0_5, 0.5_4_5_0_2, 0.5_0_6_8_0, 0.4_7_7_7_7, 0.4_1_0_2_8, 0.4_5_3_0_4]      )

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


      def    a (self       :      Dict      ):





         """simple docstring"""



         __snake_case							       =   OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint				,			provider='''CPUExecutionProvider'''      )
         __snake_case							       =   EulerAncestralDiscreteScheduler.from_config(pipe.scheduler.config      )
         pipe.set_progress_bar_config(disable=a__      )

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

         assert image.shape == (1, 128, 128, 3)
         __snake_case							       =   np.array([0.5_2_9_1_1, 0.6_0_0_0_4, 0.4_9_2_2_9, 0.4_9_8_0_5, 0.5_4_5_0_2, 0.5_0_6_8_0, 0.4_7_7_7_7, 0.4_1_0_2_8, 0.4_5_3_0_4]      )

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


      def    a (self       :      List[str]      ):





         """simple docstring"""



         __snake_case							       =   OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint				,			provider='''CPUExecutionProvider'''      )
         __snake_case							       =   DPMSolverMultistepScheduler.from_config(pipe.scheduler.config      )
         pipe.set_progress_bar_config(disable=a__      )

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

         assert image.shape == (1, 128, 128, 3)
         __snake_case							       =   np.array([0.6_5_3_3_1, 0.5_8_2_7_7, 0.4_8_2_0_4, 0.5_6_0_5_9, 0.5_3_6_6_5, 0.5_6_2_3_5, 0.5_0_9_6_9, 0.4_0_0_0_9, 0.4_6_5_5_2]      )

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






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


      @property
      def    a (self       :      List[str]      ):





         """simple docstring"""



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


      @property
      def    a (self       :      Optional[Any]      ):





         """simple docstring"""



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


      def    a (self       :      Optional[Any]      ):





         """simple docstring"""



         __snake_case							       =   load_image(
             '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
             '''/img2img/sketch-mountains-input.jpg'''      )
         __snake_case							       =   init_image.resize((768, 512)      )
         # using the PNDM scheduler by default
         __snake_case							       =   OnnxStableDiffusionImgaImgPipeline.from_pretrained(
             '''CompVis/stable-diffusion-v1-4'''				,			revision='''onnx'''				,			safety_checker=a__				,			feature_extractor=a__				,			provider=self.gpu_provider				,			sess_options=self.gpu_options				,			)
         pipe.set_progress_bar_config(disable=a__      )

         __snake_case							       =   '''A fantasy landscape, trending on artstation'''

         __snake_case							       =   np.random.RandomState(0      )
         __snake_case							       =   pipe(
             prompt=a__				,			image=a__				,			strength=0.7_5				,			guidance_scale=7.5				,			num_inference_steps=10				,			generator=a__				,			output_type='''np'''				,			)
         __snake_case							       =   output.images
         __snake_case							       =   images[0, 255:258, 383:386, -1]

         assert images.shape == (1, 512, 768, 3)
         __snake_case							       =   np.array([0.4_9_0_9, 0.5_0_5_9, 0.5_3_7_2, 0.4_6_2_3, 0.4_8_7_6, 0.5_0_4_9, 0.4_8_2_0, 0.4_9_5_6, 0.5_0_1_9]      )
         # TODO: lower the tolerance after finding the cause of onnxruntime reproducibility issues

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


      def    a (self       :      Dict      ):





         """simple docstring"""



         __snake_case							       =   load_image(
             '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
             '''/img2img/sketch-mountains-input.jpg'''      )
         __snake_case							       =   init_image.resize((768, 512)      )
         __snake_case							       =   LMSDiscreteScheduler.from_pretrained(
             '''runwayml/stable-diffusion-v1-5'''				,			subfolder='''scheduler'''				,			revision='''onnx'''      )
         __snake_case							       =   OnnxStableDiffusionImgaImgPipeline.from_pretrained(
             '''runwayml/stable-diffusion-v1-5'''				,			revision='''onnx'''				,			scheduler=a__				,			safety_checker=a__				,			feature_extractor=a__				,			provider=self.gpu_provider				,			sess_options=self.gpu_options				,			)
         pipe.set_progress_bar_config(disable=a__      )

         __snake_case							       =   '''A fantasy landscape, trending on artstation'''

         __snake_case							       =   np.random.RandomState(0      )
         __snake_case							       =   pipe(
             prompt=a__				,			image=a__				,			strength=0.7_5				,			guidance_scale=7.5				,			num_inference_steps=20				,			generator=a__				,			output_type='''np'''				,			)
         __snake_case							       =   output.images
         __snake_case							       =   images[0, 255:258, 383:386, -1]

         assert images.shape == (1, 512, 768, 3)
         __snake_case							       =   np.array([0.8_0_4_3, 0.9_2_6, 0.9_5_8_1, 0.8_1_1_9, 0.8_9_5_4, 0.9_1_3, 0.7_2_0_9, 0.7_4_6_3, 0.7_4_3_1]      )
         # TODO: lower the tolerance after finding the cause of onnxruntime reproducibility issues

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

