


'''simple docstring'''






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 						a__							(				a__    ):





							'''simple docstring'''






							def     __SCREAMING_SNAKE_CASE				(						self  )			->				Any:
														lowerCAmelCase__			  =      pa.array(TypedSequence([1, 2, 3]  )  )
														self.assertEqual(arr.type   ,       pa.intaa()  )







							def     __SCREAMING_SNAKE_CASE				(						self  )			->				str:
														with self.assertRaises(lowerCamelCase_  ):
																					lowerCAmelCase__			  =      pa.array(TypedSequence([1, 2, 3]  )   ,       type=pa.intaa()  )







							def     __SCREAMING_SNAKE_CASE				(						self  )			->				List[str]:
														with self.assertRaises(lowerCamelCase_  ):
																					lowerCAmelCase__			  =      pa.array(TypedSequence([1, 2, 3]   ,       try_type=Value('''bool'''  )   ,       type=Value('''int64'''  )  )  )







							def     __SCREAMING_SNAKE_CASE				(						self  )			->				Optional[Any]:
														lowerCAmelCase__			  =      pa.array(TypedSequence([1, 2, 3]   ,       type=Value('''int32'''  )  )  )
														self.assertEqual(arr.type   ,       pa.intaa()  )







							def     __SCREAMING_SNAKE_CASE				(						self  )			->				List[Any]:
														with self.assertRaises((TypeError, pa.lib.ArrowInvalid)  ):
																					lowerCAmelCase__			  =      pa.array(TypedSequence(['''foo''', '''bar''']   ,       type=Value('''int64'''  )  )  )







							def     __SCREAMING_SNAKE_CASE				(						self  )			->				Optional[Any]:
														lowerCAmelCase__			  =      pa.array(TypedSequence([1, 2, 3]   ,       try_type=Value('''int32'''  )  )  )
														self.assertEqual(arr.type   ,       pa.intaa()  )







							def     __SCREAMING_SNAKE_CASE				(						self  )			->				Any:
														lowerCAmelCase__			  =      pa.array(TypedSequence(['''foo''', '''bar''']   ,       try_type=Value('''int64'''  )  )  )
														self.assertEqual(arr.type   ,       pa.string()  )







							def     __SCREAMING_SNAKE_CASE				(						self  )			->				Optional[int]:
														lowerCAmelCase__			  =      pa.array(TypedSequence([[[1, 2, 3]]]   ,       type=ArrayaD((1, 3)   ,       '''int64'''  )  )  )
														self.assertEqual(arr.type   ,       ArrayaDExtensionType((1, 3)   ,       '''int64'''  )  )







							def     __SCREAMING_SNAKE_CASE				(						self  )			->				str:
														with self.assertRaises((TypeError, pa.lib.ArrowInvalid)  ):
																					lowerCAmelCase__			  =      pa.array(TypedSequence(['''foo''', '''bar''']   ,       type=ArrayaD((1, 3)   ,       '''int64'''  )  )  )







							def     __SCREAMING_SNAKE_CASE				(						self  )			->				Optional[int]:
														lowerCAmelCase__			  =      pa.array(TypedSequence([[[1, 2, 3]]]   ,       try_type=ArrayaD((1, 3)   ,       '''int64'''  )  )  )
														self.assertEqual(arr.type   ,       ArrayaDExtensionType((1, 3)   ,       '''int64'''  )  )







							def     __SCREAMING_SNAKE_CASE				(						self  )			->				Optional[Any]:
														lowerCAmelCase__			  =      pa.array(TypedSequence(['''foo''', '''bar''']   ,       try_type=ArrayaD((1, 3)   ,       '''int64'''  )  )  )
														self.assertEqual(arr.type   ,       pa.string()  )







							@require_pil
							def     __SCREAMING_SNAKE_CASE				(						self  )			->				Dict:
														import PIL.Image

														lowerCAmelCase__			  =      PIL.Image.fromarray(np.arange(10   ,       dtype=np.uinta  ).reshape(2   ,       5  )  )
														with patch(
														    '''datasets.arrow_writer.cast_to_python_objects'''   ,       side_effect=lowerCamelCase_  ) as mock_cast_to_python_objects:
																					lowerCAmelCase__			  =      pa.array(TypedSequence([{'''path''': None, '''bytes''': b'''image_bytes'''}, pil_image]   ,       type=Image()  )  )
																					lowerCAmelCase__  ,   lowerCAmelCase__			  =      mock_cast_to_python_objects.call_args_list[-1]
																					self.assertIn('''optimize_list_casting'''   ,       lowerCamelCase_  )
																					self.assertFalse(kwargs['''optimize_list_casting''']  )


def    _snake_case  (	A					,       A			)		->       str:
							lowerCAmelCase__			  =      pa.BufferReader(A			) if isinstance(A					,       pa.Buffer			) else pa.memory_map(A			)
							lowerCAmelCase__			  =      pa.ipc.open_stream(A			)
							lowerCAmelCase__			  =      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    _snake_case  (	A					,       A			)		->       Tuple:
							lowerCAmelCase__			  =      pa.BufferOutputStream()
							lowerCAmelCase__			  =      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}			)
														lowerCAmelCase__  ,   lowerCAmelCase__			  =      writer.finalize()
							assert num_examples == 2
							assert num_bytes > 0
							if not fields:
														lowerCAmelCase__			  =      {'''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    _snake_case  (	)		->       Dict:
							lowerCAmelCase__			  =      pa.BufferOutputStream()
							lowerCAmelCase__			  =      Features({'''labels''': ClassLabel(names=['''neg''', '''pos''']			)}			)
							with ArrowWriter(stream=A					,       features=A			) as writer:
														writer.write({'''labels''': 0}			)
														writer.write({'''labels''': 1}			)
														lowerCAmelCase__  ,   lowerCAmelCase__			  =      writer.finalize()
							assert num_examples == 2
							assert num_bytes > 0
							assert writer._schema == features.arrow_schema
							assert writer._schema.metadata == features.arrow_schema.metadata
							lowerCAmelCase__			  =      pa.BufferReader(output.getvalue()			)
							lowerCAmelCase__			  =      pa.ipc.open_stream(A			)
							lowerCAmelCase__			  =      f.read_all()
							lowerCAmelCase__			  =      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    _snake_case  (	A			)		->       Optional[int]:
							lowerCAmelCase__			  =      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]			)
																					lowerCAmelCase__  ,   lowerCAmelCase__			  =      writer.finalize()


@pytest.mark.parametrize('''writer_batch_size'''					,       [None, 2, 10]			)
def    _snake_case  (	A			)		->       int:
							lowerCAmelCase__			  =      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			)
																					lowerCAmelCase__  ,   lowerCAmelCase__			  =      writer.finalize()


@pytest.mark.parametrize('''writer_batch_size'''					,       [None, 2, 10]			)
def    _snake_case  (	A			)		->       int:
							lowerCAmelCase__			  =      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			)
														lowerCAmelCase__  ,   lowerCAmelCase__			  =      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    _snake_case  (	A					,       A			)		->       Dict:
							lowerCAmelCase__			  =      pa.BufferOutputStream()
							lowerCAmelCase__			  =      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''': []}			)
														lowerCAmelCase__  ,   lowerCAmelCase__			  =      writer.finalize()
							assert num_examples == 2
							assert num_bytes > 0
							if not fields:
														lowerCAmelCase__			  =      {'''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    _snake_case  (	A					,       A			)		->       int:
							lowerCAmelCase__			  =      pa.BufferOutputStream()
							lowerCAmelCase__			  =      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]}			)			)
														lowerCAmelCase__  ,   lowerCAmelCase__			  =      writer.finalize()
							assert num_examples == 2
							assert num_bytes > 0
							if not fields:
														lowerCAmelCase__			  =      {'''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    _snake_case  (	A					,       A			)		->       Any:
							lowerCAmelCase__			  =      pa.BufferOutputStream()
							lowerCAmelCase__			  =      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]}			)			)
														lowerCAmelCase__  ,   lowerCAmelCase__			  =      writer.finalize()
							assert num_examples == 2
							assert num_bytes > 0
							if not fields:
														lowerCAmelCase__			  =      {'''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    _snake_case  (	)		->       Dict:
							with tempfile.TemporaryDirectory() as tmp_dir:
														lowerCAmelCase__			  =      {'''col_1''': pa.string(), '''col_2''': pa.intaa()}
														lowerCAmelCase__			  =      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]}			)
																					lowerCAmelCase__  ,   lowerCAmelCase__			  =      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    _snake_case  (	A			)		->       str:
							if pa.types.is_list(A			):
														return get_base_dtype(arr_type.value_type			)
							else:
														return arr_type


def    _snake_case  (	A					,       A			)		->       Tuple:
							if isinstance(lst[0]					,       A			):
														change_first_primitive_element_in_list(lst[0]					,       A			)
							else:
														lowerCAmelCase__			  =      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    _snake_case  (	A					,       A					,       A			)		->       Optional[int]:
							lowerCAmelCase__			  =      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    _snake_case  (	A					,       A					,       A			)		->       Dict:
							# in range
							lowerCAmelCase__			  =      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
														lowerCAmelCase__			  =      copy.deepcopy(A			)
														lowerCAmelCase__			  =      np.iinfo(expected_dtype.to_pandas_dtype()			).max + 1
														change_first_primitive_element_in_list(A					,       A			)
														lowerCAmelCase__			  =      pa.array(OptimizedTypedSequence(A					,       col=A			)			)
														assert get_base_dtype(arr.type			) == pa.intaa()


@pytest.mark.parametrize('''raise_exception'''					,       [False, True]			)
def    _snake_case  (	A					,       A			)		->       List[str]:
							lowerCAmelCase__			  =      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    _snake_case  (	A			)		->       Optional[int]:
							lowerCAmelCase__			  =      '''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}			)
														lowerCAmelCase__  ,   lowerCAmelCase__			  =      writer.finalize()
							assert num_examples == 2
							assert num_bytes > 0
							assert mockfs.exists(A			)


def    _snake_case  (	)		->       Union[str, Any]:
							lowerCAmelCase__			  =      pa.BufferOutputStream()
							with ParquetWriter(stream=A			) as writer:
														writer.write({'''col_1''': '''foo''', '''col_2''': 1}			)
														writer.write({'''col_1''': '''bar''', '''col_2''': 2}			)
														lowerCAmelCase__  ,   lowerCAmelCase__			  =      writer.finalize()
							assert num_examples == 2
							assert num_bytes > 0
							lowerCAmelCase__			  =      pa.BufferReader(output.getvalue()			)
							lowerCAmelCase__			  =      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    _snake_case  (	A					,       A			)		->       Optional[int]:
							import PIL.Image

							lowerCAmelCase__			  =      str(tmp_path / '''test_image_rgb.jpg'''			)
							PIL.Image.fromarray(np.zeros((5, 5)					,       dtype=np.uinta			)			).save(A					,       format='''png'''			)
							lowerCAmelCase__			  =      pa.BufferOutputStream()
							with ParquetWriter(
							    stream=A					,       features=Features({'''image''': Image()}			)					,       embed_local_files=A			) as writer:
														writer.write({'''image''': image_path}			)
														writer.finalize()
							lowerCAmelCase__			  =      pa.BufferReader(output.getvalue()			)
							lowerCAmelCase__			  =      pq.read_table(A			)
							lowerCAmelCase__			  =      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    _snake_case  (	)		->       Optional[Any]:
							lowerCAmelCase__			  =      pa.schema([pa.field('''col_1'''					,       pa.string()					,       nullable=A			)]			)
							lowerCAmelCase__			  =      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 tempfile
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMInverseScheduler,
    DDIMScheduler,
    DPMSolverMultistepInverseScheduler,
    DPMSolverMultistepScheduler,
    StableDiffusionDiffEditPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import load_image, slow
from diffusers.utils.testing_utils import enable_full_determinism, floats_tensor, require_torch_gpu, torch_device

from ..pipeline_params import TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS, TEXT_GUIDED_IMAGE_INPAINTING_PARAMS
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()






class     UpperCAmelCase						(		A_			,A_			,unittest.TestCase					):
   A__   :						Dict			=	StableDiffusionDiffEditPipeline
   A__   :						str			=	TEXT_GUIDED_IMAGE_INPAINTING_PARAMS - {"height", "width", "image"} | {"image_latents"}
   A__   :						List[Any]			=	TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS - {"image"} | {"image_latents"}
   A__   :						Optional[Any]			=	frozenset(
       []					)  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
   A__   :						int			=	frozenset([]					)



   def 							_SCREAMING_SNAKE_CASE  (self						:   Tuple					)						->      Optional[Any]:


          '''simple docstring'''


          torch.manual_seed(0					)
          snake_case			:  Dict     							= UNetaDConditionModel(
              block_out_channels=(32, 64)       ,		layers_per_block=2       ,		sample_size=32       ,		in_channels=4       ,		out_channels=4       ,		down_block_types=("DownBlock2D", "CrossAttnDownBlock2D")       ,		up_block_types=("CrossAttnUpBlock2D", "UpBlock2D")       ,		cross_attention_dim=32       ,		attention_head_dim=(2, 4)       ,		use_linear_projection=snake_case__       ,		)
          snake_case			:  str     							= DDIMScheduler(
              beta_start=0.00085       ,		beta_end=0.012       ,		beta_schedule="scaled_linear"       ,		clip_sample=snake_case__       ,		set_alpha_to_one=snake_case__       ,		)
          snake_case			:  List[Any]     							= DDIMInverseScheduler(
              beta_start=0.00085       ,		beta_end=0.012       ,		beta_schedule="scaled_linear"       ,		clip_sample=snake_case__       ,		set_alpha_to_zero=snake_case__       ,		)
          torch.manual_seed(0					)
          snake_case			:  Any     							= 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=1_28       ,		)
          torch.manual_seed(0					)
          snake_case			:  Union[str, Any]     							= 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=10_00       ,		hidden_act="gelu"       ,		projection_dim=5_12       ,		)
          snake_case			:  str     							= CLIPTextModel(snake_case__					)
          snake_case			:  Optional[Any]     							= CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"					)

          snake_case			:  List[str]     							= {
              "unet": unet,
              "scheduler": scheduler,
              "inverse_scheduler": inverse_scheduler,
              "vae": vae,
              "text_encoder": text_encoder,
              "tokenizer": tokenizer,
              "safety_checker": None,
              "feature_extractor": None,
          }

          return components



   def 							_SCREAMING_SNAKE_CASE  (self						:   List[str]       ,		snake_case__						:   Dict       ,		snake_case__						:   Union[str, Any]=0					)						->      Optional[Any]:


          '''simple docstring'''


          snake_case			:  List[Any]     							= floats_tensor((1, 16, 16)       ,		rng=random.Random(snake_case__					)					).to(snake_case__					)
          snake_case			:  List[Any]     							= floats_tensor((1, 2, 4, 16, 16)       ,		rng=random.Random(snake_case__					)					).to(snake_case__					)
          if str(snake_case__					).startswith("mps"					):
                 snake_case			:  Any     							= torch.manual_seed(snake_case__					)
          else:
                 snake_case			:  Optional[Any]     							= torch.Generator(device=snake_case__					).manual_seed(snake_case__					)
          snake_case			:  Union[str, Any]     							= {
              "prompt": "a dog and a newt",
              "mask_image": mask,
              "image_latents": latents,
              "generator": generator,
              "num_inference_steps": 2,
              "inpaint_strength": 1.0,
              "guidance_scale": 6.0,
              "output_type": "numpy",
          }

          return inputs



   def 							_SCREAMING_SNAKE_CASE  (self						:   Tuple       ,		snake_case__						:   Optional[int]       ,		snake_case__						:   Optional[int]=0					)						->      Tuple:


          '''simple docstring'''


          snake_case			:  Tuple     							= floats_tensor((1, 3, 32, 32)       ,		rng=random.Random(snake_case__					)					).to(snake_case__					)
          snake_case			:  Optional[Any]     							= image.cpu().permute(0       ,		2       ,		3       ,		1					)[0]
          snake_case			:  Any     							= Image.fromarray(np.uinta(snake_case__					)					).convert("RGB"					)
          if str(snake_case__					).startswith("mps"					):
                 snake_case			:  int     							= torch.manual_seed(snake_case__					)
          else:
                 snake_case			:  Optional[int]     							= torch.Generator(device=snake_case__					).manual_seed(snake_case__					)
          snake_case			:  Any     							= {
              "image": image,
              "source_prompt": "a cat and a frog",
              "target_prompt": "a dog and a newt",
              "generator": generator,
              "num_inference_steps": 2,
              "num_maps_per_mask": 2,
              "mask_encode_strength": 1.0,
              "guidance_scale": 6.0,
              "output_type": "numpy",
          }

          return inputs



   def 							_SCREAMING_SNAKE_CASE  (self						:   Any       ,		snake_case__						:   Optional[Any]       ,		snake_case__						:   Tuple=0					)						->      Any:


          '''simple docstring'''


          snake_case			:  Optional[Any]     							= floats_tensor((1, 3, 32, 32)       ,		rng=random.Random(snake_case__					)					).to(snake_case__					)
          snake_case			:  Tuple     							= image.cpu().permute(0       ,		2       ,		3       ,		1					)[0]
          snake_case			:  List[Any]     							= Image.fromarray(np.uinta(snake_case__					)					).convert("RGB"					)
          if str(snake_case__					).startswith("mps"					):
                 snake_case			:  str     							= torch.manual_seed(snake_case__					)
          else:
                 snake_case			:  int     							= torch.Generator(device=snake_case__					).manual_seed(snake_case__					)
          snake_case			:  List[str]     							= {
              "image": image,
              "prompt": "a cat and a frog",
              "generator": generator,
              "num_inference_steps": 2,
              "inpaint_strength": 1.0,
              "guidance_scale": 6.0,
              "decode_latents": True,
              "output_type": "numpy",
          }
          return inputs



   def 							_SCREAMING_SNAKE_CASE  (self						:   List[str]					)						->      str:


          '''simple docstring'''


          if not hasattr(self.pipeline_class       ,		"_optional_components"					):
                 return

          snake_case			:  int     							= self.get_dummy_components()
          snake_case			:  Optional[int]     							= self.pipeline_class(**snake_case__					)
          pipe.to(snake_case__					)
          pipe.set_progress_bar_config(disable=snake_case__					)

          # set all optional components to None and update pipeline config accordingly
          for optional_component in pipe._optional_components:
                 setattr(snake_case__       ,		snake_case__       ,		snake_case__					)
          pipe.register_modules(**{optional_component: None for optional_component in pipe._optional_components}					)

          snake_case			:  Dict     							= self.get_dummy_inputs(snake_case__					)
          snake_case			:  int     							= pipe(**snake_case__					)[0]

          with tempfile.TemporaryDirectory() as tmpdir:
                 pipe.save_pretrained(snake_case__					)
                 snake_case			:  Any     							= self.pipeline_class.from_pretrained(snake_case__					)
                 pipe_loaded.to(snake_case__					)
                 pipe_loaded.set_progress_bar_config(disable=snake_case__					)

          for optional_component in pipe._optional_components:
                 self.assertTrue(
                     getattr(snake_case__       ,		snake_case__					) is None       ,		f"""`{optional_component}` did not stay set to None after loading."""       ,		)

          snake_case			:  Optional[int]     							= self.get_dummy_inputs(snake_case__					)
          snake_case			:  int     							= pipe_loaded(**snake_case__					)[0]

          snake_case			:  int     							= np.abs(output - output_loaded					).max()
          self.assertLess(snake_case__       ,		1e-4					)



   def 							_SCREAMING_SNAKE_CASE  (self						:   str					)						->      int:


          '''simple docstring'''


          snake_case			:  Union[str, Any]     							= "cpu"

          snake_case			:  int     							= self.get_dummy_components()
          snake_case			:  Any     							= self.pipeline_class(**snake_case__					)
          pipe.to(snake_case__					)
          pipe.set_progress_bar_config(disable=snake_case__					)

          snake_case			:  Any     							= self.get_dummy_mask_inputs(snake_case__					)
          snake_case			:  Union[str, Any]     							= pipe.generate_mask(**snake_case__					)
          snake_case			:  List[Any]     							= mask[0, -3:, -3:]

          self.assertEqual(mask.shape       ,		(1, 16, 16)					)
          snake_case			:  int     							= np.array([0] * 9					)
          snake_case			:  str     							= np.abs(mask_slice.flatten() - expected_slice					).max()
          self.assertLessEqual(snake_case__       ,		1e-3					)
          self.assertEqual(mask[0, -3, -4]       ,		0					)



   def 							_SCREAMING_SNAKE_CASE  (self						:   str					)						->      str:


          '''simple docstring'''


          snake_case			:  Any     							= "cpu"

          snake_case			:  List[Any]     							= self.get_dummy_components()
          snake_case			:  Optional[int]     							= self.pipeline_class(**snake_case__					)
          pipe.to(snake_case__					)
          pipe.set_progress_bar_config(disable=snake_case__					)

          snake_case			:  List[str]     							= self.get_dummy_inversion_inputs(snake_case__					)
          snake_case			:  List[str]     							= pipe.invert(**snake_case__					).images
          snake_case			:  Optional[Any]     							= image[0, -1, -3:, -3:]

          self.assertEqual(image.shape       ,		(2, 32, 32, 3)					)
          snake_case			:  int     							= np.array(
              [0.5150, 0.5134, 0.5043, 0.5376, 0.4694, 0.51050, 0.5015, 0.4407, 0.4799]       ,		)
          snake_case			:  Optional[Any]     							= np.abs(image_slice.flatten() - expected_slice					).max()
          self.assertLessEqual(snake_case__       ,		1e-3					)



   def 							_SCREAMING_SNAKE_CASE  (self						:   int					)						->      Any:


          '''simple docstring'''


          super().test_inference_batch_single_identical(expected_max_diff=5e-3					)



   def 							_SCREAMING_SNAKE_CASE  (self						:   str					)						->      Any:


          '''simple docstring'''


          snake_case			:  Any     							= "cpu"

          snake_case			:  int     							= self.get_dummy_components()

          snake_case			:  Tuple     							= {"beta_start": 0.00085, "beta_end": 0.012, "beta_schedule": "scaled_linear"}
          snake_case			:  int     							= DPMSolverMultistepScheduler(**snake_case__					)
          snake_case			:  Optional[Any]     							= DPMSolverMultistepInverseScheduler(**snake_case__					)

          snake_case			:  Any     							= self.pipeline_class(**snake_case__					)
          pipe.to(snake_case__					)
          pipe.set_progress_bar_config(disable=snake_case__					)

          snake_case			:  Tuple     							= self.get_dummy_inversion_inputs(snake_case__					)
          snake_case			:  Optional[int]     							= pipe.invert(**snake_case__					).images
          snake_case			:  Optional[Any]     							= image[0, -1, -3:, -3:]

          self.assertEqual(image.shape       ,		(2, 32, 32, 3)					)
          snake_case			:  int     							= np.array(
              [0.5150, 0.5134, 0.5043, 0.5376, 0.4694, 0.51050, 0.5015, 0.4407, 0.4799]       ,		)
          snake_case			:  Any     							= np.abs(image_slice.flatten() - expected_slice					).max()
          self.assertLessEqual(snake_case__       ,		1e-3					)






@require_torch_gpu
@slow
class     UpperCAmelCase						(		unittest.TestCase					):



   def 							_SCREAMING_SNAKE_CASE  (self						:   List[Any]					)						->      int:


          '''simple docstring'''


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



   @classmethod
   def 							_SCREAMING_SNAKE_CASE  (cls						:   Optional[int]					)						->      Optional[int]:


          '''simple docstring'''


          snake_case			:  Tuple     							= load_image(
              "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/diffedit/fruit.png"					)

          snake_case			:  Tuple     							= raw_image.convert("RGB"					).resize((7_68, 7_68)					)

          snake_case			:  Tuple     							= raw_image



   def 							_SCREAMING_SNAKE_CASE  (self						:   Union[str, Any]					)						->      Optional[Any]:


          '''simple docstring'''


          snake_case			:  Optional[int]     							= torch.manual_seed(0					)

          snake_case			:  List[str]     							= StableDiffusionDiffEditPipeline.from_pretrained(
              "stabilityai/stable-diffusion-2-1"       ,		safety_checker=snake_case__       ,		torch_dtype=torch.floataa					)
          snake_case			:  int     							= DDIMScheduler.from_config(pipe.scheduler.config					)
          snake_case			:  Dict     							= DDIMInverseScheduler.from_config(pipe.scheduler.config					)
          pipe.enable_model_cpu_offload()
          pipe.set_progress_bar_config(disable=snake_case__					)

          snake_case			:  Union[str, Any]     							= "a bowl of fruit"
          snake_case			:  int     							= "a bowl of pears"

          snake_case			:  List[Any]     							= pipe.generate_mask(
              image=self.raw_image       ,		source_prompt=snake_case__       ,		target_prompt=snake_case__       ,		generator=snake_case__       ,		)

          snake_case			:  Optional[int]     							= pipe.invert(
              prompt=snake_case__       ,		image=self.raw_image       ,		inpaint_strength=0.7       ,		generator=snake_case__					).latents

          snake_case			:  List[str]     							= pipe(
              prompt=snake_case__       ,		mask_image=snake_case__       ,		image_latents=snake_case__       ,		generator=snake_case__       ,		negative_prompt=snake_case__       ,		inpaint_strength=0.7       ,		output_type="numpy"       ,		).images[0]

          snake_case			:  Dict     							= (
              np.array(
                  load_image(
                      "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
                      "/diffedit/pears.png"					).resize((7_68, 7_68)					)					)
              / 2_55
          )
          assert np.abs((expected_image - image).max()					) < 5e-1



   def 							_SCREAMING_SNAKE_CASE  (self						:   Union[str, Any]					)						->      Tuple:


          '''simple docstring'''


          snake_case			:  int     							= torch.manual_seed(0					)

          snake_case			:  Tuple     							= StableDiffusionDiffEditPipeline.from_pretrained(
              "stabilityai/stable-diffusion-2-1"       ,		safety_checker=snake_case__       ,		torch_dtype=torch.floataa					)
          snake_case			:  Union[str, Any]     							= DPMSolverMultistepScheduler.from_config(pipe.scheduler.config					)
          snake_case			:  Optional[int]     							= DPMSolverMultistepInverseScheduler.from_config(pipe.scheduler.config					)
          pipe.enable_model_cpu_offload()
          pipe.set_progress_bar_config(disable=snake_case__					)

          snake_case			:  str     							= "a bowl of fruit"
          snake_case			:  Optional[int]     							= "a bowl of pears"

          snake_case			:  Optional[Any]     							= pipe.generate_mask(
              image=self.raw_image       ,		source_prompt=snake_case__       ,		target_prompt=snake_case__       ,		generator=snake_case__       ,		)

          snake_case			:  List[str]     							= pipe.invert(
              prompt=snake_case__       ,		image=self.raw_image       ,		inpaint_strength=0.7       ,		generator=snake_case__       ,		num_inference_steps=25       ,		).latents

          snake_case			:  Dict     							= pipe(
              prompt=snake_case__       ,		mask_image=snake_case__       ,		image_latents=snake_case__       ,		generator=snake_case__       ,		negative_prompt=snake_case__       ,		inpaint_strength=0.7       ,		num_inference_steps=25       ,		output_type="numpy"       ,		).images[0]

          snake_case			:  int     							= (
              np.array(
                  load_image(
                      "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
                      "/diffedit/pears.png"					).resize((7_68, 7_68)					)					)
              / 2_55
          )
          assert np.abs((expected_image - image).max()					) < 5e-1





