"""simple docstring"""






import random
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    UNetaDConditionModel,
    VideoToVideoSDPipeline,
)
from diffusers.utils import floats_tensor, is_xformers_available, skip_mps
from diffusers.utils.testing_utils import enable_full_determinism, slow, torch_device

from ..pipeline_params import (
    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()





@skip_mps
class 				_A		( lowerCAmelCase						,					unittest.TestCase			):
      snake_case__    :     Optional[int]													=			VideoToVideoSDPipeline
      snake_case__    :     Union[str, Any]													=			TEXT_GUIDED_IMAGE_VARIATION_PARAMS.union({'video'}			) - {'image', 'width', 'height'}
      snake_case__    :     Union[str, Any]													=			TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({'video'}			) - {'image'}
      snake_case__    :     Dict													=			PipelineTesterMixin.required_optional_params - {'latents'}
      snake_case__    :     Any													=			False

      # No `output_type`.
      snake_case__    :     Dict													=			frozenset(
          [
              'num_inference_steps',
              'generator',
              'latents',
              'return_dict',
              'callback',
              'callback_steps',
          ]			)

      def        A__ (		self		):







            """simple docstring"""



            torch.manual_seed(0		)
            lowercase														=			UNetaDConditionModel(
                block_out_channels=(32, 64, 64, 64)	, layers_per_block=2	, sample_size=32	, in_channels=4	, out_channels=4	, down_block_types=("""CrossAttnDownBlock3D""", """CrossAttnDownBlock3D""", """CrossAttnDownBlock3D""", """DownBlock3D""")	, up_block_types=("""UpBlock3D""", """CrossAttnUpBlock3D""", """CrossAttnUpBlock3D""", """CrossAttnUpBlock3D""")	, cross_attention_dim=32	, attention_head_dim=4	, )
            lowercase														=			DDIMScheduler(
                beta_start=0.0_0_0_8_5	, beta_end=0.0_1_2	, beta_schedule="""scaled_linear"""	, clip_sample=__lowerCAmelCase	, set_alpha_to_one=__lowerCAmelCase	, )
            torch.manual_seed(0		)
            lowercase														=			AutoencoderKL(
                block_out_channels=[32, 64]	, in_channels=3	, out_channels=3	, down_block_types=["""DownEncoderBlock2D""", """DownEncoderBlock2D"""]	, up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D"""]	, latent_channels=4	, sample_size=128	, )
            torch.manual_seed(0		)
            lowercase														=			CLIPTextConfig(
                bos_token_id=0	, eos_token_id=2	, hidden_size=32	, intermediate_size=37	, layer_norm_eps=1E-05	, num_attention_heads=4	, num_hidden_layers=5	, pad_token_id=1	, vocab_size=1000	, hidden_act="""gelu"""	, projection_dim=512	, )
            lowercase														=			CLIPTextModel(__lowerCAmelCase		)
            lowercase														=			CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""		)

            lowercase														=			{
                """unet""": unet,
                """scheduler""": scheduler,
                """vae""": vae,
                """text_encoder""": text_encoder,
                """tokenizer""": tokenizer,
            }
            return components

      def        A__ (		self	, __lowerCAmelCase	, __lowerCAmelCase=0		):







            """simple docstring"""



            lowercase														=			floats_tensor((1, 3, 3, 32, 32)	, rng=random.Random(__lowerCAmelCase		)		).to(__lowerCAmelCase		)

            if str(__lowerCAmelCase		).startswith("""mps"""		):
                  lowercase														=			torch.manual_seed(__lowerCAmelCase		)
            else:
                  lowercase														=			torch.Generator(device=__lowerCAmelCase		).manual_seed(__lowerCAmelCase		)
            lowercase														=			{
                """prompt""": """A painting of a squirrel eating a burger""",
                """video""": video,
                """generator""": generator,
                """num_inference_steps""": 2,
                """guidance_scale""": 6.0,
                """output_type""": """pt""",
            }
            return inputs

      def        A__ (		self		):







            """simple docstring"""



            lowercase														=			"""cpu"""  # ensure determinism for the device-dependent torch.Generator
            lowercase														=			self.get_dummy_components()
            lowercase														=			VideoToVideoSDPipeline(**__lowerCAmelCase		)
            lowercase														=			sd_pipe.to(__lowerCAmelCase		)
            sd_pipe.set_progress_bar_config(disable=__lowerCAmelCase		)

            lowercase														=			self.get_dummy_inputs(__lowerCAmelCase		)
            lowercase														=			"""np"""
            lowercase														=			sd_pipe(**__lowerCAmelCase		).frames
            lowercase														=			frames[0][-3:, -3:, -1]

            assert frames[0].shape == (32, 32, 3)
            lowercase														=			np.array([106, 117, 113, 174, 137, 112, 148, 151, 131]		)

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

      @unittest.skipIf(
          torch_device != """cuda""" or not is_xformers_available()	, reason="""XFormers attention is only available with CUDA and `xformers` installed"""	, )
      def        A__ (		self		):







            """simple docstring"""



            self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=__lowerCAmelCase	, expected_max_diff=5E-3		)

      @unittest.skip(reason="""Batching needs to be properly figured out first for this pipeline."""		)
      def        A__ (		self		):







            """simple docstring"""



            pass

      @unittest.skip(reason="""Batching needs to be properly figured out first for this pipeline."""		)
      def        A__ (		self		):







            """simple docstring"""



            pass

      @unittest.skip(reason="""`num_images_per_prompt` argument is not supported for this pipeline."""		)
      def        A__ (		self		):







            """simple docstring"""



            pass







      def        A__ (		self		):







            """simple docstring"""



            return super().test_progress_bar()






@slow
@skip_mps
class 				_A		( unittest.TestCase			):

      def        A__ (		self		):







            """simple docstring"""



            lowercase														=			VideoToVideoSDPipeline.from_pretrained("""cerspense/zeroscope_v2_XL"""	, torch_dtype=torch.floataa		)
            pipe.enable_model_cpu_offload()

            # 10 frames
            lowercase														=			torch.Generator(device="""cpu"""		).manual_seed(0		)
            lowercase														=			torch.randn((1, 10, 3, 1024, 576)	, generator=__lowerCAmelCase		)
            lowercase														=			video.to("""cuda"""		)

            lowercase														=			"""Spiderman is surfing"""

            lowercase														=			pipe(__lowerCAmelCase	, video=__lowerCAmelCase	, generator=__lowerCAmelCase	, num_inference_steps=3	, output_type="""pt"""		).frames

            lowercase														=			np.array([-1.0_4_5_8_9_8_4, -1.1_2_7_9_2_9_7, -0.9_6_6_3_0_8_6, -0.9_1_5_0_3_9_0_6, -0.7_5_0_9_7_6_5_6]		)
            assert np.abs(video_frames.cpu().numpy()[0, 0, 0, 0, -5:] - expected_array		).sum() < 1E-2




"""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()



