






import os
from typing import Optional

import fsspec
from fsspec.archive import AbstractArchiveFileSystem
from fsspec.utils import DEFAULT_BLOCK_SIZE

class        lowerCAmelCase_				(      lowercase	):







				"""simple docstring"""




				_snake_case				:   Dict        =			""""""
				_snake_case				:   str        =			(
				    None  # protocol passed in prefix to the url. ex: "gzip", for gzip://file.txt::http://foo.bar/file.txt.gz
				)
				_snake_case				:   str        =			None  # compression type in fsspec. ex: "gzip"
				_snake_case				:   str        =			None  # extension of the filename to strip. ex: "".gz" to get file.txt from file.txt.gz







				def __init__( self     :Optional[Any]  ,     lowerCamelCase__     :str = ""  ,     lowerCamelCase__     :Optional[str] = None  ,     lowerCamelCase__     :Optional[dict] = None  ,     **lowerCamelCase__     :Dict     ):
					super().__init__(self  ,     **lowerCamelCase__     )
					# always open as "rb" since fsspec can then use the TextIOWrapper to make it work for "r" mode
					UpperCamelCase__ :Optional[int]						=      fsspec.open(
					    lowerCamelCase__  ,     mode="""rb"""  ,     protocol=lowerCamelCase__  ,     compression=self.compression  ,     client_kwargs={
					        """requote_redirect_url""": False,  # see https://github.com/huggingface/datasets/pull/5459
					        """trust_env""": True,  # Enable reading proxy env variables.
					        **(target_options or {}).pop("""client_kwargs"""  ,     {}     ),  # To avoid issues if it was already passed.
					    }  ,     **(target_options or {})  ,     )
					UpperCamelCase__ :str						=      os.path.basename(self.file.path.split("""::"""     )[0]     )
					UpperCamelCase__ :List[Any]						=      (
					    self.compressed_name[: self.compressed_name.rindex("""."""     )]
					    if """.""" in self.compressed_name
					    else self.compressed_name
					)
					UpperCamelCase__ :Any						=      None







				@classmethod
				def 				__a   ( cls     :Any  ,     lowerCamelCase__     :Union[str, Any]     ):
					# compressed file paths are always relative to the archive root
					return super()._strip_protocol(lowerCamelCase__     ).lstrip("""/"""     )







				def 				__a   ( self     :List[Any]     ):
					if self.dir_cache is None:
						UpperCamelCase__ :List[str]						=      {**self.file.fs.info(self.file.path     ), """name""": self.uncompressed_name}
						UpperCamelCase__ :Tuple						=      {f["""name"""]: f}







				def 				__a   ( self     :List[Any]  ,     lowerCamelCase__     :str     ):
					return self.file.open().read()







				def 				__a   ( self     :str  ,     lowerCamelCase__     :str  ,     lowerCamelCase__     :str = "rb"  ,     lowerCamelCase__     :List[str]=None  ,     lowerCamelCase__     :Any=True  ,     lowerCamelCase__     :str=None  ,     **lowerCamelCase__     :List[str]  ,     ):
					UpperCamelCase__ :Any						=      self._strip_protocol(lowerCamelCase__     )
					if mode != "rb":
						raise ValueError(f"""Tried to read with mode {mode} on file {self.file.path} opened with mode 'rb'"""     )
					return self.file.open()

class        lowerCAmelCase_				(      lowercase	):







				"""simple docstring"""




				_snake_case				:   str        =			"""bz2"""
				_snake_case				:   Union[str, Any]        =			"""bz2"""
				_snake_case				:   Tuple        =			""".bz2"""

class        lowerCAmelCase_				(      lowercase	):







				"""simple docstring"""




				_snake_case				:   int        =			"""gzip"""
				_snake_case				:   Tuple        =			"""gzip"""
				_snake_case				:   Optional[int]        =			""".gz"""

class        lowerCAmelCase_				(      lowercase	):







				"""simple docstring"""




				_snake_case				:   Union[str, Any]        =			"""lz4"""
				_snake_case				:   str        =			"""lz4"""
				_snake_case				:   Tuple        =			""".lz4"""

class        lowerCAmelCase_				(      lowercase	):







				"""simple docstring"""




				_snake_case				:   Optional[int]        =			"""xz"""
				_snake_case				:   Union[str, Any]        =			"""xz"""
				_snake_case				:   Optional[Any]        =			""".xz"""

class        lowerCAmelCase_				(      lowercase	):







				"""simple docstring"""




				_snake_case				:   Tuple        =			"""zstd"""
				_snake_case				:   Optional[int]        =			"""zstd"""
				_snake_case				:   Dict        =			""".zst"""







				def __init__( self     :Optional[int]  ,     lowerCamelCase__     :str  ,     lowerCamelCase__     :str = "rb"  ,     lowerCamelCase__     :Optional[str] = None  ,     lowerCamelCase__     :Optional[dict] = None  ,     lowerCamelCase__     :int = DEFAULT_BLOCK_SIZE  ,     **lowerCamelCase__     :Dict  ,     ):
					super().__init__(
					    fo=lowerCamelCase__  ,     mode=lowerCamelCase__  ,     target_protocol=lowerCamelCase__  ,     target_options=lowerCamelCase__  ,     block_size=lowerCamelCase__  ,     **lowerCamelCase__  ,     )
					# We need to wrap the zstd decompressor to avoid this error in fsspec==2021.7.0 and zstandard==0.15.2:
					#
					# File "/Users/user/.virtualenvs/hf-datasets/lib/python3.7/site-packages/fsspec/core.py", line 145, in open
					#     out.close = close
					# AttributeError: 'zstd.ZstdDecompressionReader' object attribute 'close' is read-only
					#
					# see https://github.com/intake/filesystem_spec/issues/725
					UpperCamelCase__ :int						=      self.file.__enter__

					class        lowerCAmelCase_				:







									"""simple docstring"""




									def __init__( self     :Optional[Any]  ,     lowerCamelCase__     :Union[str, Any]     ):
										UpperCamelCase__ :int						=      file_







									def __enter__( self     :str     ):
										self._file.__enter__()
										return self







									def __exit__( self     :int  ,     *lowerCamelCase__     :int  ,     **lowerCamelCase__     :Dict     ):
										self._file.__exit__(*lowerCamelCase__  ,     **lowerCamelCase__     )







									def __iter__( self     :Union[str, Any]     ):
										return iter(self._file     )







									def 				__a   ( self     :Union[str, Any]     ):
										return next(self._file     )







									def __getattr__( self     :str  ,     lowerCamelCase__     :Any     ):
										return getattr(self._file  ,     lowerCamelCase__     )

					def fixed_enter(*lowerCamelCase__     :Optional[Any]  ,     **lowerCamelCase__     :int     ):
						return WrappedFile(_enter(*lowerCamelCase__  ,     **lowerCamelCase__     )     )

					UpperCamelCase__ :List[Any]						=      fixed_enter




"""simple docstring"""

import gc
import unittest

import numpy as np
import torch

from diffusers import StableDiffusionKDiffusionPipeline
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu


enable_full_determinism()
@slow
@require_torch_gpu
class     lowerCAmelCase__						(	unittest.TestCase ):


	'''simple docstring'''







	def 			_lowerCAmelCase     (      self				:       List[Any]    )						->	str:





					"""simple docstring"""


					super().tearDown()
					gc.collect()
					torch.cuda.empty_cache()





	def 			_lowerCAmelCase     (      self				:       Any    )						->	str:





					"""simple docstring"""


					SCREAMING_SNAKE_CASE						:							int							=     StableDiffusionKDiffusionPipeline.from_pretrained('CompVis/stable-diffusion-v1-4'    )
					SCREAMING_SNAKE_CASE						:							Dict							=     sd_pipe.to(_SCREAMING_SNAKE_CASE    )
					sd_pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE    )

					sd_pipe.set_scheduler('sample_euler'    )

					SCREAMING_SNAKE_CASE						:							Union[str, Any]							=     'A painting of a squirrel eating a burger'
					SCREAMING_SNAKE_CASE						:							Optional[Any]							=     torch.manual_seed(0    )
					SCREAMING_SNAKE_CASE						:							Optional[Any]							=     sd_pipe([prompt]     ,       generator=_SCREAMING_SNAKE_CASE     ,       guidance_scale=9.0     ,       num_inference_steps=20     ,       output_type='np'    )

					SCREAMING_SNAKE_CASE						:							str							=     output.images

					SCREAMING_SNAKE_CASE						:							Union[str, Any]							=     image[0, -3:, -3:, -1]

					assert image.shape == (1, 512, 512, 3)
					SCREAMING_SNAKE_CASE						:							Union[str, Any]							=     np.array([0.0_4_4_7, 0.0_4_9_2, 0.0_4_6_8, 0.0_4_0_8, 0.0_3_8_3, 0.0_4_0_8, 0.0_3_5_4, 0.0_3_8_0, 0.0_3_3_9]    )

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





	def 			_lowerCAmelCase     (      self				:       Dict    )						->	Optional[int]:





					"""simple docstring"""


					SCREAMING_SNAKE_CASE						:							Optional[Any]							=     StableDiffusionKDiffusionPipeline.from_pretrained('stabilityai/stable-diffusion-2-1-base'    )
					SCREAMING_SNAKE_CASE						:							List[Any]							=     sd_pipe.to(_SCREAMING_SNAKE_CASE    )
					sd_pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE    )

					sd_pipe.set_scheduler('sample_euler'    )

					SCREAMING_SNAKE_CASE						:							List[str]							=     'A painting of a squirrel eating a burger'
					SCREAMING_SNAKE_CASE						:							Optional[int]							=     torch.manual_seed(0    )
					SCREAMING_SNAKE_CASE						:							Dict							=     sd_pipe([prompt]     ,       generator=_SCREAMING_SNAKE_CASE     ,       guidance_scale=9.0     ,       num_inference_steps=20     ,       output_type='np'    )

					SCREAMING_SNAKE_CASE						:							Optional[int]							=     output.images

					SCREAMING_SNAKE_CASE						:							List[str]							=     image[0, -3:, -3:, -1]

					assert image.shape == (1, 512, 512, 3)
					SCREAMING_SNAKE_CASE						:							List[Any]							=     np.array([0.1_2_3_7, 0.1_3_2_0, 0.1_4_3_8, 0.1_3_5_9, 0.1_3_9_0, 0.1_1_3_2, 0.1_2_7_7, 0.1_1_7_5, 0.1_1_1_2]    )

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







	def 			_lowerCAmelCase     (      self				:       Dict    )						->	Optional[Any]:





					"""simple docstring"""


					SCREAMING_SNAKE_CASE						:							Optional[int]							=     StableDiffusionKDiffusionPipeline.from_pretrained('stabilityai/stable-diffusion-2-1-base'    )
					SCREAMING_SNAKE_CASE						:							Union[str, Any]							=     sd_pipe.to(_SCREAMING_SNAKE_CASE    )
					sd_pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE    )

					sd_pipe.set_scheduler('sample_dpmpp_2m'    )

					SCREAMING_SNAKE_CASE						:							Optional[int]							=     'A painting of a squirrel eating a burger'
					SCREAMING_SNAKE_CASE						:							str							=     torch.manual_seed(0    )
					SCREAMING_SNAKE_CASE						:							Optional[Any]							=     sd_pipe(
					    [prompt]     ,       generator=_SCREAMING_SNAKE_CASE     ,       guidance_scale=7.5     ,       num_inference_steps=15     ,       output_type='np'     ,       use_karras_sigmas=_SCREAMING_SNAKE_CASE     ,       )

					SCREAMING_SNAKE_CASE						:							int							=     output.images

					SCREAMING_SNAKE_CASE						:							List[Any]							=     image[0, -3:, -3:, -1]

					assert image.shape == (1, 512, 512, 3)
					SCREAMING_SNAKE_CASE						:							int							=     np.array(
					    [0.1_1_3_8_1_6_8_9, 0.1_2_1_1_2_9_2_1, 0.1_3_8_9_4_5_7, 0.1_2_5_4_9_6_0_6, 0.1_2_4_4_9_6_4, 0.1_0_8_3_1_5_1_7, 0.1_1_5_6_2_8_6_6, 0.1_0_8_6_7_8_1_6, 0.1_0_4_9_9_0_4_8]    )

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





