"""simple docstring"""






import gc
import tempfile
import unittest

import numpy as np
import torch

from diffusers import VersatileDiffusionPipeline
from diffusers.utils.testing_utils import load_image, nightly, require_torch_gpu, torch_device


__lowerCAmelCase					: List[str]        =False





class 				_A		( unittest.TestCase			):
      pass






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

      def        A__ (		self		):







            """simple docstring"""



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

      def        A__ (		self		):







            """simple docstring"""



            lowercase														=			VersatileDiffusionPipeline.from_pretrained("""shi-labs/versatile-diffusion"""	, torch_dtype=torch.floataa		)
            pipe.to(__lowerCAmelCase		)
            pipe.set_progress_bar_config(disable=__lowerCAmelCase		)

            lowercase														=			load_image(
                """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/versatile_diffusion/benz.jpg"""		)

            lowercase														=			torch.manual_seed(0		)
            lowercase														=			pipe.dual_guided(
                prompt="""first prompt"""	, image=__lowerCAmelCase	, text_to_image_strength=0.7_5	, generator=__lowerCAmelCase	, guidance_scale=7.5	, num_inference_steps=2	, output_type="""numpy"""	, ).images

            with tempfile.TemporaryDirectory() as tmpdirname:
                  pipe.save_pretrained(__lowerCAmelCase		)
                  lowercase														=			VersatileDiffusionPipeline.from_pretrained(__lowerCAmelCase	, torch_dtype=torch.floataa		)
            pipe.to(__lowerCAmelCase		)
            pipe.set_progress_bar_config(disable=__lowerCAmelCase		)

            lowercase														=			generator.manual_seed(0		)
            lowercase														=			pipe.dual_guided(
                prompt="""first prompt"""	, image=__lowerCAmelCase	, text_to_image_strength=0.7_5	, generator=__lowerCAmelCase	, 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        A__ (		self		):







            """simple docstring"""



            lowercase														=			VersatileDiffusionPipeline.from_pretrained("""shi-labs/versatile-diffusion"""	, torch_dtype=torch.floataa		)
            pipe.to(__lowerCAmelCase		)
            pipe.set_progress_bar_config(disable=__lowerCAmelCase		)

            lowercase														=			"""cyberpunk 2077"""
            lowercase														=			load_image(
                """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/versatile_diffusion/benz.jpg"""		)
            lowercase														=			torch.manual_seed(0		)
            lowercase														=			pipe.dual_guided(
                prompt=__lowerCAmelCase	, image=__lowerCAmelCase	, text_to_image_strength=0.7_5	, generator=__lowerCAmelCase	, guidance_scale=7.5	, num_inference_steps=50	, output_type="""numpy"""	, ).images

            lowercase														=			image[0, 253:256, 253:256, -1]

            assert image.shape == (1, 512, 512, 3)
            lowercase														=			np.array([0.1_4_4_8, 0.1_6_1_9, 0.1_7_4_1, 0.1_0_8_6, 0.1_1_4_7, 0.1_1_2_8, 0.1_1_9_9, 0.1_1_6_5, 0.1_0_0_1]		)

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

            lowercase														=			"""A painting of a squirrel eating a burger """
            lowercase														=			torch.manual_seed(0		)
            lowercase														=			pipe.text_to_image(
                prompt=__lowerCAmelCase	, generator=__lowerCAmelCase	, guidance_scale=7.5	, num_inference_steps=50	, output_type="""numpy"""		).images

            lowercase														=			image[0, 253:256, 253:256, -1]

            assert image.shape == (1, 512, 512, 3)
            lowercase														=			np.array([0.3_3_6_7, 0.3_1_6_9, 0.2_6_5_6, 0.3_8_7_0, 0.4_7_9_0, 0.3_7_9_6, 0.4_0_0_9, 0.4_8_7_8, 0.4_7_7_8]		)

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

            lowercase														=			pipe.image_variation(__lowerCAmelCase	, generator=__lowerCAmelCase	, output_type="""numpy"""		).images

            lowercase														=			image[0, 253:256, 253:256, -1]

            assert image.shape == (1, 512, 512, 3)
            lowercase														=			np.array([0.3_0_7_6, 0.3_1_2_3, 0.3_2_8_4, 0.3_7_8_2, 0.3_7_7_0, 0.3_8_9_4, 0.4_2_9_7, 0.4_3_3_1, 0.4_4_5_6]		)

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







"""simple docstring"""






import contextlib
import os
import sqlitea

import pytest

from datasets import Dataset, Features, Value
from datasets.io.sql import SqlDatasetReader, SqlDatasetWriter

from ..utils import assert_arrow_memory_doesnt_increase, assert_arrow_memory_increases, require_sqlalchemy


def     UpperCAmelCase__ (       lowerCAmelCase__  :int      ,	lowerCAmelCase__  :Tuple	)				->					Union[str, Any]:
      '''simple docstring'''


      assert isinstance(lowerCAmelCase__      ,	lowerCAmelCase__	)
      assert dataset.num_rows == 4
      assert dataset.num_columns == 3
      assert dataset.column_names == ["col_1", "col_2", "col_3"]
      for feature, expected_dtype in expected_features.items():
            assert dataset.features[feature].dtype == expected_dtype


@require_sqlalchemy
@pytest.mark.parametrize("""keep_in_memory"""      ,	[False, True]	)
def     UpperCAmelCase__ (       lowerCAmelCase__  :Dict      ,	lowerCAmelCase__  :Optional[Any]      ,	lowerCAmelCase__  :str      ,	lowerCAmelCase__  :List[str]	)				->					int:
      '''simple docstring'''


      lowercase														=			tmp_path / """cache"""
      lowercase														=			{"""col_1""": """string""", """col_2""": """int64""", """col_3""": """float64"""}
      with assert_arrow_memory_increases() if keep_in_memory else assert_arrow_memory_doesnt_increase():
            lowercase														=			SqlDatasetReader(
                """dataset"""      ,	"""sqlite:///""" + sqlite_path      ,	cache_dir=lowerCAmelCase__      ,	keep_in_memory=lowerCAmelCase__	).read()
      _check_sql_dataset(lowerCAmelCase__      ,	lowerCAmelCase__	)


@require_sqlalchemy
@pytest.mark.parametrize(
    """features"""      ,	[
        None,
        {"""col_1""": """string""", """col_2""": """int64""", """col_3""": """float64"""},
        {"""col_1""": """string""", """col_2""": """string""", """col_3""": """string"""},
        {"""col_1""": """int32""", """col_2""": """int32""", """col_3""": """int32"""},
        {"""col_1""": """float32""", """col_2""": """float32""", """col_3""": """float32"""},
    ]      ,	)
def     UpperCAmelCase__ (       lowerCAmelCase__  :int      ,	lowerCAmelCase__  :Union[str, Any]      ,	lowerCAmelCase__  :List[Any]      ,	lowerCAmelCase__  :Union[str, Any]	)				->					Optional[int]:
      '''simple docstring'''


      lowercase														=			tmp_path / """cache"""
      lowercase														=			{"""col_1""": """string""", """col_2""": """int64""", """col_3""": """float64"""}
      lowercase														=			features.copy() if features else default_expected_features
      lowercase														=			(
          Features({feature: Value(lowerCAmelCase__	) for feature, dtype in features.items()}	) if features is not None else None
      )
      lowercase														=			SqlDatasetReader("""dataset"""      ,	"""sqlite:///""" + sqlite_path      ,	features=lowerCAmelCase__      ,	cache_dir=lowerCAmelCase__	).read()
      _check_sql_dataset(lowerCAmelCase__      ,	lowerCAmelCase__	)


def     UpperCAmelCase__ (       lowerCAmelCase__  :Any	)				->					Optional[int]:
      '''simple docstring'''


      with contextlib.closing(sqlitea.connect(lowerCAmelCase__	)	) as con:
            lowercase														=			con.cursor()
            cur.execute("""SELECT * FROM dataset"""	)
            for row in cur:
                  yield row


@require_sqlalchemy
def     UpperCAmelCase__ (       lowerCAmelCase__  :Tuple      ,	lowerCAmelCase__  :str      ,	lowerCAmelCase__  :int	)				->					List[Any]:
      '''simple docstring'''


      lowercase														=			tmp_path / """cache"""
      lowercase														=			os.path.join(lowerCAmelCase__      ,	"""tmp.sql"""	)
      lowercase														=			SqlDatasetReader("""dataset"""      ,	"""sqlite:///""" + sqlite_path      ,	cache_dir=lowerCAmelCase__	).read()
      SqlDatasetWriter(lowerCAmelCase__      ,	"""dataset"""      ,	"""sqlite:///""" + output_sqlite_path      ,	num_proc=1	).write()

      lowercase														=			iter_sql_file(lowerCAmelCase__	)
      lowercase														=			iter_sql_file(lowerCAmelCase__	)

      for rowa, rowa in zip(lowerCAmelCase__      ,	lowerCAmelCase__	):
            assert rowa == rowa


@require_sqlalchemy
def     UpperCAmelCase__ (       lowerCAmelCase__  :str      ,	lowerCAmelCase__  :Optional[Any]      ,	lowerCAmelCase__  :Optional[int]	)				->					Optional[int]:
      '''simple docstring'''


      lowercase														=			tmp_path / """cache"""
      lowercase														=			os.path.join(lowerCAmelCase__      ,	"""tmp.sql"""	)
      lowercase														=			SqlDatasetReader("""dataset"""      ,	"""sqlite:///""" + sqlite_path      ,	cache_dir=lowerCAmelCase__	).read()
      SqlDatasetWriter(lowerCAmelCase__      ,	"""dataset"""      ,	"""sqlite:///""" + output_sqlite_path      ,	num_proc=2	).write()

      lowercase														=			iter_sql_file(lowerCAmelCase__	)
      lowercase														=			iter_sql_file(lowerCAmelCase__	)

      for rowa, rowa in zip(lowerCAmelCase__      ,	lowerCAmelCase__	):
            assert rowa == rowa






@require_sqlalchemy
def     UpperCAmelCase__ (       lowerCAmelCase__  :Dict      ,	lowerCAmelCase__  :Dict      ,	lowerCAmelCase__  :Tuple	)				->					Any:
      '''simple docstring'''


      lowercase														=			tmp_path / """cache"""
      lowercase														=			os.path.join(lowerCAmelCase__      ,	"""tmp.sql"""	)
      lowercase														=			SqlDatasetReader("""dataset"""      ,	"""sqlite:///""" + sqlite_path      ,	cache_dir=lowerCAmelCase__	).read()
      with pytest.raises(lowerCAmelCase__	):
            SqlDatasetWriter(lowerCAmelCase__      ,	"""dataset"""      ,	"""sqlite:///""" + output_sqlite_path      ,	num_proc=0	).write()



