






import copy
import os
import tempfile
from unittest import TestCase
from unittest.mock import patch

import numpy as np
import pyarrow as pa
import pyarrow.parquet as pq
import pytest

from datasets.arrow_writer import ArrowWriter, OptimizedTypedSequence, ParquetWriter, TypedSequence
from datasets.features import ArrayaD, ClassLabel, Features, Image, Value
from datasets.features.features import ArrayaDExtensionType, cast_to_python_objects
from datasets.keyhash import DuplicatedKeysError, InvalidKeyError

from .utils import require_pil







class  _lowercase   (   UpperCAmelCase_						):
      def 						lowercase__					(     self   ):
       snake_case__  :    Any							       =pa.array(TypedSequence([1, 2, 3]   )   )
       self.assertEqual(arr.type				,			pa.intaa()   )
      def 						lowercase__					(     self   ):
       with self.assertRaises(_lowercase   ):
        snake_case__  :    str							       =pa.array(TypedSequence([1, 2, 3]   )				,			type=pa.intaa()   )
      def 						lowercase__					(     self   ):
       with self.assertRaises(_lowercase   ):
        snake_case__  :    Optional[int]							       =pa.array(TypedSequence([1, 2, 3]				,			try_type=Value("""bool"""   )				,			type=Value("""int64"""   )   )   )
      def 						lowercase__					(     self   ):
       snake_case__  :    Any							       =pa.array(TypedSequence([1, 2, 3]				,			type=Value("""int32"""   )   )   )
       self.assertEqual(arr.type				,			pa.intaa()   )
      def 						lowercase__					(     self   ):
       with self.assertRaises((TypeError, pa.lib.ArrowInvalid)   ):
        snake_case__  :    Union[str, Any]							       =pa.array(TypedSequence(["""foo""", """bar"""]				,			type=Value("""int64"""   )   )   )
      def 						lowercase__					(     self   ):
       snake_case__  :    Dict							       =pa.array(TypedSequence([1, 2, 3]				,			try_type=Value("""int32"""   )   )   )
       self.assertEqual(arr.type				,			pa.intaa()   )
      def 						lowercase__					(     self   ):
       snake_case__  :    List[Any]							       =pa.array(TypedSequence(["""foo""", """bar"""]				,			try_type=Value("""int64"""   )   )   )
       self.assertEqual(arr.type				,			pa.string()   )
      def 						lowercase__					(     self   ):
       snake_case__  :    Dict							       =pa.array(TypedSequence([[[1, 2, 3]]]				,			type=ArrayaD((1, 3)				,			"""int64"""   )   )   )
       self.assertEqual(arr.type				,			ArrayaDExtensionType((1, 3)				,			"""int64"""   )   )
      def 						lowercase__					(     self   ):
       with self.assertRaises((TypeError, pa.lib.ArrowInvalid)   ):
        snake_case__  :    List[Any]							       =pa.array(TypedSequence(["""foo""", """bar"""]				,			type=ArrayaD((1, 3)				,			"""int64"""   )   )   )
      def 						lowercase__					(     self   ):
       snake_case__  :    Tuple							       =pa.array(TypedSequence([[[1, 2, 3]]]				,			try_type=ArrayaD((1, 3)				,			"""int64"""   )   )   )
       self.assertEqual(arr.type				,			ArrayaDExtensionType((1, 3)				,			"""int64"""   )   )
      def 						lowercase__					(     self   ):
       snake_case__  :    Optional[Any]							       =pa.array(TypedSequence(["""foo""", """bar"""]				,			try_type=ArrayaD((1, 3)				,			"""int64"""   )   )   )
       self.assertEqual(arr.type				,			pa.string()   )



      @require_pil
      def 						lowercase__					(     self   ):
       import PIL.Image

       snake_case__  :    Optional[Any]							       =PIL.Image.fromarray(np.arange(1_0				,			dtype=np.uinta   ).reshape(2				,			5   )   )
       with patch(
           """datasets.arrow_writer.cast_to_python_objects"""				,			side_effect=_lowercase   ) as mock_cast_to_python_objects:
        snake_case__  :    List[Any]							       =pa.array(TypedSequence([{"""path""": None, """bytes""": B"""image_bytes"""}, pil_image]				,			type=Image()   )   )
        snake_case__  :    Any							       =mock_cast_to_python_objects.call_args_list[-1]
        self.assertIn("""optimize_list_casting"""				,			_lowercase   )
        self.assertFalse(kwargs["""optimize_list_casting"""]   )







def 			A__							(      _a  :					List[Any]		,  _a  :					int	):





 '''simple docstring'''







 snake_case__  :    Union[str, Any]							       =pa.BufferReader(_a	) if isinstance(_a		,  pa.Buffer	) else pa.memory_map(_a	)
 snake_case__  :    Tuple							       =pa.ipc.open_stream(_a	)
 snake_case__  :    pa.Table							       =f.read_all()
 assert len(pa_table.to_batches()	) == expected_num_chunks
 assert pa_table.to_pydict() == {"col_1": ["foo", "bar"], "col_2": [1, 2]}
 del pa_table







@pytest.mark.parametrize("""writer_batch_size"""		,  [None, 1, 10]	)
@pytest.mark.parametrize(
    """fields"""		,  [None, {"""col_1""": pa.string(), """col_2""": pa.intaa()}, {"""col_1""": pa.string(), """col_2""": pa.intaa()}]	)
def 			A__							(      _a  :					str		,  _a  :					int	):





 '''simple docstring'''







 snake_case__  :    str							       =pa.BufferOutputStream()
 snake_case__  :    Union[str, Any]							       =pa.schema(_a	) if fields else None
 with ArrowWriter(stream=_a		,  schema=_a		,  writer_batch_size=_a	) as writer:
  writer.write({"""col_1""": """foo""", """col_2""": 1}	)
  writer.write({"""col_1""": """bar""", """col_2""": 2}	)
  snake_case__  :    str							       =writer.finalize()
 assert num_examples == 2
 assert num_bytes > 0
 if not fields:
  snake_case__  :    str							       ={'col_1': pa.string(), 'col_2': pa.intaa()}
 assert writer._schema == pa.schema(_a		,  metadata=writer._schema.metadata	)
 _check_output(output.getvalue()		,  expected_num_chunks=num_examples if writer_batch_size == 1 else 1	)







def 			A__							(      ):





 '''simple docstring'''







 snake_case__  :    Any							       =pa.BufferOutputStream()
 snake_case__  :    Dict							       =Features({"""labels""": ClassLabel(names=["""neg""", """pos"""]	)}	)
 with ArrowWriter(stream=_a		,  features=_a	) as writer:
  writer.write({"""labels""": 0}	)
  writer.write({"""labels""": 1}	)
  snake_case__  :    Optional[int]							       =writer.finalize()
 assert num_examples == 2
 assert num_bytes > 0
 assert writer._schema == features.arrow_schema
 assert writer._schema.metadata == features.arrow_schema.metadata
 snake_case__  :    Dict							       =pa.BufferReader(output.getvalue()	)
 snake_case__  :    str							       =pa.ipc.open_stream(_a	)
 snake_case__  :    pa.Table							       =f.read_all()
 snake_case__  :    Optional[int]							       =pa_table.schema
 assert pa_table.num_rows == 2
 assert schema == features.arrow_schema
 assert schema.metadata == features.arrow_schema.metadata
 assert features == Features.from_arrow_schema(_a	)







@pytest.mark.parametrize("""writer_batch_size"""		,  [None, 1, 10]	)
def 			A__							(      _a  :					int	):





 '''simple docstring'''







 snake_case__  :    Union[str, Any]							       =pa.BufferOutputStream()
 with ArrowWriter(
     stream=_a		,  writer_batch_size=_a		,  hash_salt="""split_name"""		,  check_duplicates=_a		,  ) as writer:
  with pytest.raises(_a	):
   writer.write({"""col_1""": """foo""", """col_2""": 1}		,  key=[1, 2]	)
   snake_case__  :    Optional[int]							       =writer.finalize()







@pytest.mark.parametrize("""writer_batch_size"""		,  [None, 2, 10]	)
def 			A__							(      _a  :					Tuple	):





 '''simple docstring'''







 snake_case__  :    List[str]							       =pa.BufferOutputStream()
 with ArrowWriter(
     stream=_a		,  writer_batch_size=_a		,  hash_salt="""split_name"""		,  check_duplicates=_a		,  ) as writer:
  with pytest.raises(_a	):
   writer.write({"""col_1""": """foo""", """col_2""": 1}		,  key=10	)
   writer.write({"""col_1""": """bar""", """col_2""": 2}		,  key=10	)
   snake_case__  :    Optional[Any]							       =writer.finalize()







@pytest.mark.parametrize("""writer_batch_size"""		,  [None, 2, 10]	)
def 			A__							(      _a  :					Union[str, Any]	):





 '''simple docstring'''







 snake_case__  :    int							       =pa.BufferOutputStream()
 with ArrowWriter(
     stream=_a		,  writer_batch_size=_a		,  hash_salt="""split_name"""		,  check_duplicates=_a		,  ) as writer:
  writer.write({"""col_1""": """foo""", """col_2""": 1}		,  key=1	)
  writer.write({"""col_1""": """bar""", """col_2""": 2}		,  key=2	)
  snake_case__  :    Any							       =writer.finalize()
 assert num_examples == 2
 assert num_bytes > 0
 _check_output(output.getvalue()		,  expected_num_chunks=num_examples if writer_batch_size == 1 else 1	)







@pytest.mark.parametrize("""writer_batch_size"""		,  [None, 1, 10]	)
@pytest.mark.parametrize(
    """fields"""		,  [None, {"""col_1""": pa.string(), """col_2""": pa.intaa()}, {"""col_1""": pa.string(), """col_2""": pa.intaa()}]	)
def 			A__							(      _a  :					Optional[Any]		,  _a  :					Optional[Any]	):





 '''simple docstring'''







 snake_case__  :    List[str]							       =pa.BufferOutputStream()
 snake_case__  :    Dict							       =pa.schema(_a	) if fields else None
 with ArrowWriter(stream=_a		,  schema=_a		,  writer_batch_size=_a	) as writer:
  writer.write_batch({"""col_1""": ["""foo""", """bar"""], """col_2""": [1, 2]}	)
  writer.write_batch({"""col_1""": [], """col_2""": []}	)
  snake_case__  :    str							       =writer.finalize()
 assert num_examples == 2
 assert num_bytes > 0
 if not fields:
  snake_case__  :    str							       ={'col_1': pa.string(), 'col_2': pa.intaa()}
 assert writer._schema == pa.schema(_a		,  metadata=writer._schema.metadata	)
 _check_output(output.getvalue()		,  expected_num_chunks=num_examples if writer_batch_size == 1 else 1	)







@pytest.mark.parametrize("""writer_batch_size"""		,  [None, 1, 10]	)
@pytest.mark.parametrize(
    """fields"""		,  [None, {"""col_1""": pa.string(), """col_2""": pa.intaa()}, {"""col_1""": pa.string(), """col_2""": pa.intaa()}]	)
def 			A__							(      _a  :					Dict		,  _a  :					str	):





 '''simple docstring'''







 snake_case__  :    Optional[Any]							       =pa.BufferOutputStream()
 snake_case__  :    List[Any]							       =pa.schema(_a	) if fields else None
 with ArrowWriter(stream=_a		,  schema=_a		,  writer_batch_size=_a	) as writer:
  writer.write_table(pa.Table.from_pydict({"""col_1""": ["""foo""", """bar"""], """col_2""": [1, 2]}	)	)
  snake_case__  :    int							       =writer.finalize()
 assert num_examples == 2
 assert num_bytes > 0
 if not fields:
  snake_case__  :    Union[str, Any]							       ={'col_1': pa.string(), 'col_2': pa.intaa()}
 assert writer._schema == pa.schema(_a		,  metadata=writer._schema.metadata	)
 _check_output(output.getvalue()		,  expected_num_chunks=num_examples if writer_batch_size == 1 else 1	)







@pytest.mark.parametrize("""writer_batch_size"""		,  [None, 1, 10]	)
@pytest.mark.parametrize(
    """fields"""		,  [None, {"""col_1""": pa.string(), """col_2""": pa.intaa()}, {"""col_1""": pa.string(), """col_2""": pa.intaa()}]	)
def 			A__							(      _a  :					List[Any]		,  _a  :					int	):





 '''simple docstring'''







 snake_case__  :    Tuple							       =pa.BufferOutputStream()
 snake_case__  :    Tuple							       =pa.schema(_a	) if fields else None
 with ArrowWriter(stream=_a		,  schema=_a		,  writer_batch_size=_a	) as writer:
  writer.write_row(pa.Table.from_pydict({"""col_1""": ["""foo"""], """col_2""": [1]}	)	)
  writer.write_row(pa.Table.from_pydict({"""col_1""": ["""bar"""], """col_2""": [2]}	)	)
  snake_case__  :    int							       =writer.finalize()
 assert num_examples == 2
 assert num_bytes > 0
 if not fields:
  snake_case__  :    Dict							       ={'col_1': pa.string(), 'col_2': pa.intaa()}
 assert writer._schema == pa.schema(_a		,  metadata=writer._schema.metadata	)
 _check_output(output.getvalue()		,  expected_num_chunks=num_examples if writer_batch_size == 1 else 1	)







def 			A__							(      ):





 '''simple docstring'''







 with tempfile.TemporaryDirectory() as tmp_dir:
  snake_case__  :    Optional[int]							       ={'col_1': pa.string(), 'col_2': pa.intaa()}
  snake_case__  :    Dict							       =os.path.join(_a		,  """test.arrow"""	)
  with ArrowWriter(path=_a		,  schema=pa.schema(_a	)	) as writer:
   writer.write_batch({"""col_1""": ["""foo""", """bar"""], """col_2""": [1, 2]}	)
   snake_case__  :    str							       =writer.finalize()
  assert num_examples == 2
  assert num_bytes > 0
  assert writer._schema == pa.schema(_a		,  metadata=writer._schema.metadata	)
  _check_output(_a		,  1	)







def 			A__							(      _a  :					Optional[int]	):





 '''simple docstring'''







 if pa.types.is_list(_a	):
  return get_base_dtype(arr_type.value_type	)
 else:
  return arr_type







def 			A__							(      _a  :					List[str]		,  _a  :					str	):





 '''simple docstring'''







 if isinstance(lst[0]		,  _a	):
  change_first_primitive_element_in_list(lst[0]		,  _a	)
 else:
  snake_case__  :    str							       =value







@pytest.mark.parametrize("""optimized_int_type, expected_dtype"""		,  [(None, pa.intaa()), (Value("""int32"""	), pa.intaa())]	)
@pytest.mark.parametrize("""sequence"""		,  [[1, 2, 3], [[1, 2, 3]], [[[1, 2, 3]]]]	)
def 			A__							(      _a  :					Dict		,  _a  :					Tuple		,  _a  :					Dict	):





 '''simple docstring'''







 snake_case__  :    Any							       =pa.array(TypedSequence(_a		,  optimized_int_type=_a	)	)
 assert get_base_dtype(arr.type	) == expected_dtype







@pytest.mark.parametrize(
    """col, expected_dtype"""		,  [
        ("""attention_mask""", pa.inta()),
        ("""special_tokens_mask""", pa.inta()),
        ("""token_type_ids""", pa.inta()),
        ("""input_ids""", pa.intaa()),
        ("""other""", pa.intaa()),
    ]		,  )
@pytest.mark.parametrize("""sequence"""		,  [[1, 2, 3], [[1, 2, 3]], [[[1, 2, 3]]]]	)
def 			A__							(      _a  :					Optional[Any]		,  _a  :					List[Any]		,  _a  :					List[Any]	):





 '''simple docstring'''







 snake_case__  :    int							       =pa.array(OptimizedTypedSequence(_a		,  col=_a	)	)
 assert get_base_dtype(arr.type	) == expected_dtype

 # not in range
 if col != "other":
  # avoids errors due to in-place modifications
  snake_case__  :    Union[str, Any]							       =copy.deepcopy(_a	)
  snake_case__  :    str							       =np.iinfo(expected_dtype.to_pandas_dtype()	).max + 1
  change_first_primitive_element_in_list(_a		,  _a	)
  snake_case__  :    Tuple							       =pa.array(OptimizedTypedSequence(_a		,  col=_a	)	)
  assert get_base_dtype(arr.type	) == pa.intaa()







@pytest.mark.parametrize("""raise_exception"""		,  [False, True]	)
def 			A__							(      _a  :					int		,  _a  :					Optional[int]	):





 '''simple docstring'''







 snake_case__  :    Any							       =str(tmp_path / """dataset-train.arrow"""	)
 try:
  with ArrowWriter(path=_a	) as writer:
   if raise_exception:
    raise pa.lib.ArrowInvalid()
   else:
    writer.stream.close()
 except pa.lib.ArrowInvalid:
  pass
 finally:
  assert writer.stream.closed







def 			A__							(      _a  :					Dict	):





 '''simple docstring'''







 snake_case__  :    Tuple							       ='mock://dataset-train.arrow'
 with ArrowWriter(path=_a		,  storage_options=mockfs.storage_options	) as writer:
  assert isinstance(writer._fs		,  type(_a	)	)
  assert writer._fs.storage_options == mockfs.storage_options
  writer.write({"""col_1""": """foo""", """col_2""": 1}	)
  writer.write({"""col_1""": """bar""", """col_2""": 2}	)
  snake_case__  :    Optional[Any]							       =writer.finalize()
 assert num_examples == 2
 assert num_bytes > 0
 assert mockfs.exists(_a	)







def 			A__							(      ):





 '''simple docstring'''







 snake_case__  :    Tuple							       =pa.BufferOutputStream()
 with ParquetWriter(stream=_a	) as writer:
  writer.write({"""col_1""": """foo""", """col_2""": 1}	)
  writer.write({"""col_1""": """bar""", """col_2""": 2}	)
  snake_case__  :    str							       =writer.finalize()
 assert num_examples == 2
 assert num_bytes > 0
 snake_case__  :    Union[str, Any]							       =pa.BufferReader(output.getvalue()	)
 snake_case__  :    pa.Table							       =pq.read_table(_a	)
 assert pa_table.to_pydict() == {"col_1": ["foo", "bar"], "col_2": [1, 2]}







@require_pil
@pytest.mark.parametrize("""embed_local_files"""		,  [False, True]	)
def 			A__							(      _a  :					Optional[int]		,  _a  :					Optional[Any]	):





 '''simple docstring'''







 import PIL.Image

 snake_case__  :    str							       =str(tmp_path / """test_image_rgb.jpg"""	)
 PIL.Image.fromarray(np.zeros((5, 5)		,  dtype=np.uinta	)	).save(_a		,  format="""png"""	)
 snake_case__  :    Optional[Any]							       =pa.BufferOutputStream()
 with ParquetWriter(
     stream=_a		,  features=Features({"""image""": Image()}	)		,  embed_local_files=_a	) as writer:
  writer.write({"""image""": image_path}	)
  writer.finalize()
 snake_case__  :    Optional[int]							       =pa.BufferReader(output.getvalue()	)
 snake_case__  :    pa.Table							       =pq.read_table(_a	)
 snake_case__  :    List[Any]							       =pa_table.to_pydict()
 if embed_local_files:
  assert isinstance(out["""image"""][0]["""path"""]		,  _a	)
  with open(_a		,  """rb"""	) as f:
   assert out["image"][0]["bytes"] == f.read()
 else:
  assert out["image"][0]["path"] == image_path
  assert out["image"][0]["bytes"] is None







def 			A__							(      ):





 '''simple docstring'''







 snake_case__  :    int							       =pa.schema([pa.field("""col_1"""		,  pa.string()		,  nullable=_a	)]	)
 snake_case__  :    List[str]							       =pa.BufferOutputStream()
 with ArrowWriter(stream=_a	) as writer:
  writer._build_writer(inferred_schema=_a	)
  assert writer._schema == pa.schema([pa.field("""col_1"""		,  pa.string()	)]	)













import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import XLMRobertaTokenizerFast

from diffusers import DDIMScheduler, KandinskyImgaImgPipeline, KandinskyPriorPipeline, UNetaDConditionModel, VQModel
from diffusers.pipelines.kandinsky.text_encoder import MCLIPConfig, MultilingualCLIP
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()







class  _lowercase   (   _A						,				unittest.TestCase						):
      _a							:						Any										=							KandinskyImgaImgPipeline
      _a							:						List[Any]										=							['prompt', 'image_embeds', 'negative_image_embeds', 'image']
      _a							:						int										=							[
          'prompt',
          'negative_prompt',
          'image_embeds',
          'negative_image_embeds',
          'image',
      ]
      _a							:						int										=							[
          'generator',
          'height',
          'width',
          'strength',
          'guidance_scale',
          'negative_prompt',
          'num_inference_steps',
          'return_dict',
          'guidance_scale',
          'num_images_per_prompt',
          'output_type',
          'return_dict',
      ]
      _a							:						List[str]										=							False
      @property
      def 						lowercase__					(     self   ):
       return 3_2
      @property
      def 						lowercase__					(     self   ):
       return 3_2
      @property
      def 						lowercase__					(     self   ):
       return self.time_input_dim
      @property
      def 						lowercase__					(     self   ):
       return self.time_input_dim * 4
      @property
      def 						lowercase__					(     self   ):
       return 1_0_0
      @property
      def 						lowercase__					(     self   ):
       snake_case__  :    Dict							       =XLMRobertaTokenizerFast.from_pretrained("""YiYiXu/tiny-random-mclip-base"""   )
       return tokenizer
      @property
      def 						lowercase__					(     self   ):
       torch.manual_seed(0   )
       snake_case__  :    Optional[Any]							       =MCLIPConfig(
           numDims=self.cross_attention_dim				,			transformerDimensions=self.text_embedder_hidden_size				,			hidden_size=self.text_embedder_hidden_size				,			intermediate_size=3_7				,			num_attention_heads=4				,			num_hidden_layers=5				,			vocab_size=1_0_0_5				,			)

       snake_case__  :    Optional[int]							       =MultilingualCLIP(a   )
       snake_case__  :    Union[str, Any]							       =text_encoder.eval()

       return text_encoder
      @property
      def 						lowercase__					(     self   ):
       torch.manual_seed(0   )

       snake_case__  :    Optional[int]							       ={
           """in_channels""": 4,
           # Out channels is double in channels because predicts mean and variance
           """out_channels""": 8,
           """addition_embed_type""": """text_image""",
           """down_block_types""": ("""ResnetDownsampleBlock2D""", """SimpleCrossAttnDownBlock2D"""),
           """up_block_types""": ("""SimpleCrossAttnUpBlock2D""", """ResnetUpsampleBlock2D"""),
           """mid_block_type""": """UNetMidBlock2DSimpleCrossAttn""",
           """block_out_channels""": (self.block_out_channels_a, self.block_out_channels_a * 2),
           """layers_per_block""": 1,
           """encoder_hid_dim""": self.text_embedder_hidden_size,
           """encoder_hid_dim_type""": """text_image_proj""",
           """cross_attention_dim""": self.cross_attention_dim,
           """attention_head_dim""": 4,
           """resnet_time_scale_shift""": """scale_shift""",
           """class_embed_type""": None,
       }

       snake_case__  :    List[str]							       =UNetaDConditionModel(**a   )
       return model
      @property
      def 						lowercase__					(     self   ):
       return {
           "block_out_channels": [3_2, 6_4],
           "down_block_types": ["DownEncoderBlock2D", "AttnDownEncoderBlock2D"],
           "in_channels": 3,
           "latent_channels": 4,
           "layers_per_block": 1,
           "norm_num_groups": 8,
           "norm_type": "spatial",
           "num_vq_embeddings": 1_2,
           "out_channels": 3,
           "up_block_types": [
               "AttnUpDecoderBlock2D",
               "UpDecoderBlock2D",
           ],
           "vq_embed_dim": 4,
       }
      @property
      def 						lowercase__					(     self   ):
       torch.manual_seed(0   )
       snake_case__  :    List[Any]							       =VQModel(**self.dummy_movq_kwargs   )
       return model
      def 						lowercase__					(     self   ):
       snake_case__  :    str							       =self.dummy_text_encoder
       snake_case__  :    Tuple							       =self.dummy_tokenizer
       snake_case__  :    List[str]							       =self.dummy_unet
       snake_case__  :    Dict							       =self.dummy_movq

       snake_case__  :    Union[str, Any]							       ={
           """num_train_timesteps""": 1_0_0_0,
           """beta_schedule""": """linear""",
           """beta_start""": 0.00085,
           """beta_end""": 0.012,
           """clip_sample""": False,
           """set_alpha_to_one""": False,
           """steps_offset""": 0,
           """prediction_type""": """epsilon""",
           """thresholding""": False,
       }

       snake_case__  :    Union[str, Any]							       =DDIMScheduler(**a   )

       snake_case__  :    Any							       ={
           """text_encoder""": text_encoder,
           """tokenizer""": tokenizer,
           """unet""": unet,
           """scheduler""": scheduler,
           """movq""": movq,
       }

       return components
      def 						lowercase__					(     self				,			a				,			a=0   ):
       snake_case__  :    str							       =floats_tensor((1, self.cross_attention_dim)				,			rng=random.Random(a   )   ).to(a   )
       snake_case__  :    Tuple							       =floats_tensor((1, self.cross_attention_dim)				,			rng=random.Random(seed + 1   )   ).to(a   )
       # create init_image
       snake_case__  :    str							       =floats_tensor((1, 3, 6_4, 6_4)				,			rng=random.Random(a   )   ).to(a   )
       snake_case__  :    Tuple							       =image.cpu().permute(0				,			2				,			3				,			1   )[0]
       snake_case__  :    Dict							       =Image.fromarray(np.uinta(a   )   ).convert("""RGB"""   ).resize((2_5_6, 2_5_6)   )

       if str(a   ).startswith("""mps"""   ):
        snake_case__  :    Dict							       =torch.manual_seed(a   )
       else:
        snake_case__  :    Optional[Any]							       =torch.Generator(device=a   ).manual_seed(a   )
       snake_case__  :    str							       ={
           """prompt""": """horse""",
           """image""": init_image,
           """image_embeds""": image_embeds,
           """negative_image_embeds""": negative_image_embeds,
           """generator""": generator,
           """height""": 6_4,
           """width""": 6_4,
           """num_inference_steps""": 1_0,
           """guidance_scale""": 7.0,
           """strength""": 0.2,
           """output_type""": """np""",
       }
       return inputs



      def 						lowercase__					(     self   ):
       snake_case__  :    List[str]							       ="""cpu"""

       snake_case__  :    Tuple							       =self.get_dummy_components()

       snake_case__  :    Optional[int]							       =self.pipeline_class(**a   )
       snake_case__  :    Union[str, Any]							       =pipe.to(a   )

       pipe.set_progress_bar_config(disable=a   )

       snake_case__  :    List[Any]							       =pipe(**self.get_dummy_inputs(a   )   )
       snake_case__  :    List[str]							       =output.images

       snake_case__  :    int							       =pipe(
           **self.get_dummy_inputs(a   )				,			return_dict=a				,			)[0]

       snake_case__  :    str							       =image[0, -3:, -3:, -1]
       snake_case__  :    Tuple							       =image_from_tuple[0, -3:, -3:, -1]

       assert image.shape == (1, 6_4, 6_4, 3)

       snake_case__  :    List[str]							       =np.array(
           [0.61474943, 0.6073539, 0.43308544, 0.5928269, 0.47493595, 0.46755973, 0.4613838, 0.45368797, 0.50119233]   )
       assert (
           np.abs(image_slice.flatten() - expected_slice   ).max() < 1e-2
       ), F" expected_slice {expected_slice}, but got {image_slice.flatten()}"
       assert (
           np.abs(image_from_tuple_slice.flatten() - expected_slice   ).max() < 1e-2
       ), F" expected_slice {expected_slice}, but got {image_from_tuple_slice.flatten()}"







@slow
@require_torch_gpu
class  _lowercase   (   unittest.TestCase						):
      def 						lowercase__					(     self   ):
       # clean up the VRAM after each test
       super().tearDown()
       gc.collect()
       torch.cuda.empty_cache()



      def 						lowercase__					(     self   ):
       snake_case__  :    Any							       =load_numpy(
           """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
           """/kandinsky/kandinsky_img2img_frog.npy"""   )

       snake_case__  :    Dict							       =load_image(
           """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main""" """/kandinsky/cat.png"""   )
       snake_case__  :    Any							       ="""A red cartoon frog, 4k"""

       snake_case__  :    Dict							       =KandinskyPriorPipeline.from_pretrained(
           """kandinsky-community/kandinsky-2-1-prior"""				,			torch_dtype=torch.floataa   )
       pipe_prior.to(a   )

       snake_case__  :    Tuple							       =KandinskyImgaImgPipeline.from_pretrained(
           """kandinsky-community/kandinsky-2-1"""				,			torch_dtype=torch.floataa   )
       snake_case__  :    str							       =pipeline.to(a   )

       pipeline.set_progress_bar_config(disable=a   )

       snake_case__  :    Optional[Any]							       =torch.Generator(device="""cpu"""   ).manual_seed(0   )
       snake_case__       ,   snake_case__  :    Any							       =pipe_prior(
           a				,			generator=a				,			num_inference_steps=5				,			negative_prompt=""""""				,			).to_tuple()

       snake_case__  :    str							       =pipeline(
           a				,			image=a				,			image_embeds=a				,			negative_image_embeds=a				,			generator=a				,			num_inference_steps=1_0_0				,			height=7_6_8				,			width=7_6_8				,			strength=0.2				,			output_type="""np"""				,			)

       snake_case__  :    Optional[int]							       =output.images[0]

       assert image.shape == (7_6_8, 7_6_8, 3)

       assert_mean_pixel_difference(a				,			a   )





