


import os
from pathlib import Path
from unittest.mock import patch

import pytest
import zstandard as zstd

from datasets.download.download_config import DownloadConfig
from datasets.utils.file_utils import (
    OfflineModeIsEnabled,
    cached_path,
    fsspec_get,
    fsspec_head,
    ftp_get,
    ftp_head,
    get_from_cache,
    http_get,
    http_head,
)


snake_case__			 =						'''\
    Text data.
    Second line of data.'''

snake_case__			 =						'''file'''


@pytest.fixture(scope="session"				)
def 				lowerCamelCase__	(      a	:   Dict				)    ->	Union[str, Any]:





			"""simple docstring"""
			a__						  =    tmp_path_factory.mktemp("data"				) / (FILE_PATH + ".zstd")
			a__						  =    bytes(a	,  "utf-8"				)
			with zstd.open(a	,  "wb"				) as f:
						f.write(a				)
			return path


@pytest.fixture
def 				lowerCamelCase__	(      a	:   Any				)    ->	str:





			"""simple docstring"""
			with open(os.path.join(tmpfs.local_root_dir	,  a				)	,  "w"				) as f:
						f.write(a				)
			return FILE_PATH


@pytest.mark.parametrize("compression_format"	,  ["gzip", "xz", "zstd"]				)
def 				lowerCamelCase__	(      a	:   str	,  a	:   Optional[int]	,  a	:   Optional[Any]	,  a	:   int	,  a	:   List[str]	,  a	:   List[Any]				)    ->	str:





			"""simple docstring"""
			a__						  =    {"gzip": gz_file, "xz": xz_file, "zstd": zstd_path}
			a__						  =    input_paths[compression_format]
			a__						  =    tmp_path / "cache"
			a__						  =    DownloadConfig(cache_dir=a	,  extract_compressed_file=a				)
			a__						  =    cached_path(a	,  download_config=a				)
			with open(a				) as f:
						a__						  =    f.read()
			with open(a				) as f:
						a__						  =    f.read()
			assert extracted_file_content == expected_file_content


@pytest.mark.parametrize("default_extracted"	,  [True, False]				)
@pytest.mark.parametrize("default_cache_dir"	,  [True, False]				)
def 				lowerCamelCase__	(      a	:   int	,  a	:   List[Any]	,  a	:   List[Any]	,  a	:   List[str]	,  a	:   List[str]				)    ->	Optional[int]:





			"""simple docstring"""
			a__						  =    "custom_cache"
			a__						  =    "custom_extracted_dir"
			a__						  =    tmp_path / "custom_extracted_path"
			if default_extracted:
						a__						  =    ("downloads" if default_cache_dir else custom_cache_dir, "extracted")
			else:
						monkeypatch.setattr("datasets.config.EXTRACTED_DATASETS_DIR"	,  a				)
						monkeypatch.setattr("datasets.config.EXTRACTED_DATASETS_PATH"	,  str(a				)				)
						a__						  =    custom_extracted_path.parts[-2:] if default_cache_dir else (custom_cache_dir, custom_extracted_dir)

			a__						  =    xz_file
			a__						  =    (
			    DownloadConfig(extract_compressed_file=a				)
			    if default_cache_dir
			    else DownloadConfig(cache_dir=tmp_path / custom_cache_dir	,  extract_compressed_file=a				)
			)
			a__						  =    cached_path(a	,  download_config=a				)
			assert Path(a				).parent.parts[-2:] == expected


def 				lowerCamelCase__	(      a	:   int				)    ->	Dict:





			"""simple docstring"""
			a__						  =    str(Path(a				).resolve()				)
			assert cached_path(a				) == text_file
			# relative path
			a__						  =    str(Path(a				).resolve().relative_to(Path(os.getcwd()				)				)				)
			assert cached_path(a				) == text_file


def 				lowerCamelCase__	(      a	:   List[Any]				)    ->	Optional[int]:





			"""simple docstring"""
			a__						  =    str(tmp_path.resolve() / "__missing_file__.txt"				)
			with pytest.raises(a				):
						cached_path(a				)
			# relative path
			a__						  =    "./__missing_file__.txt"
			with pytest.raises(a				):
						cached_path(a				)


def 				lowerCamelCase__	(      a	:   Optional[int]				)    ->	Optional[Any]:





			"""simple docstring"""
			a__						  =    get_from_cache(F'''tmp://{tmpfs_file}'''				)
			with open(a				) as f:
						a__						  =    f.read()
			assert output_file_content == FILE_CONTENT


@patch("datasets.config.HF_DATASETS_OFFLINE"	,  a				)
def 				lowerCamelCase__	(      )    ->	Optional[Any]:





			"""simple docstring"""
			with pytest.raises(a				):
						cached_path("https://huggingface.co"				)


@patch("datasets.config.HF_DATASETS_OFFLINE"	,  a				)
def 				lowerCamelCase__	(      a	:   str				)    ->	int:





			"""simple docstring"""
			a__						  =    tmp_path_factory.mktemp("data"				) / "file.html"
			with pytest.raises(a				):
						http_get("https://huggingface.co"	,  temp_file=a				)
			with pytest.raises(a				):
						http_head("https://huggingface.co"				)


@patch("datasets.config.HF_DATASETS_OFFLINE"	,  a				)
def 				lowerCamelCase__	(      a	:   Dict				)    ->	List[str]:





			"""simple docstring"""
			a__						  =    tmp_path_factory.mktemp("data"				) / "file.html"
			with pytest.raises(a				):
						ftp_get("ftp://huggingface.co"	,  temp_file=a				)
			with pytest.raises(a				):
						ftp_head("ftp://huggingface.co"				)







@patch("datasets.config.HF_DATASETS_OFFLINE"	,  a				)
def 				lowerCamelCase__	(      a	:   List[Any]				)    ->	Optional[Any]:





			"""simple docstring"""
			a__						  =    tmp_path_factory.mktemp("data"				) / "file.html"
			with pytest.raises(a				):
						fsspec_get("s3://huggingface.co"	,  temp_file=a				)
			with pytest.raises(a				):
						fsspec_head("s3://huggingface.co"				)






import gc
import tempfile
import unittest

import numpy as np
import torch

from diffusers import VersatileDiffusionTextToImagePipeline
from diffusers.utils.testing_utils import nightly, require_torch_gpu, torch_device


snake_case__			 =						False


class 					lowerCAmelCase_ (       unittest.TestCase):
						pass






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




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






									"""simple docstring"""






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




						def  _snake_case					(	self						:   Dict						)      ->Any:






									"""simple docstring"""






									a__ :Union[str, Any]						  =    VersatileDiffusionTextToImagePipeline.from_pretrained("shi-labs/versatile-diffusion"						)
									# remove text_unet
									pipe.remove_unused_weights()
									pipe.to(__A						)
									pipe.set_progress_bar_config(disable=__A						)

									a__ :List[Any]						  =    "A painting of a squirrel eating a burger "
									a__ :Optional[Any]						  =    torch.manual_seed(0						)
									a__ :List[Any]						  =    pipe(
									    prompt=__A   ,     generator=__A   ,     guidance_scale=7.5   ,     num_inference_steps=2   ,     output_type="numpy"						).images

									with tempfile.TemporaryDirectory() as tmpdirname:
												pipe.save_pretrained(__A						)
												a__ :List[Any]						  =    VersatileDiffusionTextToImagePipeline.from_pretrained(__A						)
									pipe.to(__A						)
									pipe.set_progress_bar_config(disable=__A						)

									a__ :Optional[int]						  =    generator.manual_seed(0						)
									a__ :List[Any]						  =    pipe(
									    prompt=__A   ,     generator=__A   ,     guidance_scale=7.5   ,     num_inference_steps=2   ,     output_type="numpy"						).images

									assert np.abs(image - new_image						).sum() < 1E-5, "Models don't have the same forward pass"






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






									"""simple docstring"""






									a__ :Tuple						  =    VersatileDiffusionTextToImagePipeline.from_pretrained(
									    "shi-labs/versatile-diffusion"   ,     torch_dtype=torch.floataa						)
									pipe.to(__A						)
									pipe.set_progress_bar_config(disable=__A						)

									a__ :Tuple						  =    "A painting of a squirrel eating a burger "
									a__ :Tuple						  =    torch.manual_seed(0						)
									a__ :Optional[Any]						  =    pipe(
									    prompt=__A   ,     generator=__A   ,     guidance_scale=7.5   ,     num_inference_steps=50   ,     output_type="numpy"						).images

									a__ :Tuple						  =    image[0, 253:256, 253:256, -1]

									assert image.shape == (1, 512, 512, 3)
									a__ :Tuple						  =    np.array([0.3_367, 0.3_169, 0.2_656, 0.3_870, 0.4_790, 0.3_796, 0.4_009, 0.4_878, 0.4_778]						)

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


