import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import CLIPImageProcessor, CLIPVisionConfig, CLIPVisionModelWithProjection

from diffusers import (
    AutoencoderKL,
    DPMSolverMultistepScheduler,
    PNDMScheduler,
    StableDiffusionImageVariationPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import floats_tensor, load_image, load_numpy, nightly, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import IMAGE_VARIATION_BATCH_PARAMS, IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()





class   A							(       UpperCAmelCase_	,       UpperCAmelCase_	,       UpperCAmelCase_	,       unittest.TestCase  ):
   __UpperCAmelCase				:				Optional[int]          =    StableDiffusionImageVariationPipeline
   __UpperCAmelCase				:				Optional[int]          =    IMAGE_VARIATION_PARAMS
   __UpperCAmelCase				:				str          =    IMAGE_VARIATION_BATCH_PARAMS
   __UpperCAmelCase				:				Optional[int]          =    frozenset([]  )
   # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
   __UpperCAmelCase				:				List[str]          =    frozenset([]  )


   def     lowercase_   (self  :		Optional[int]	)		->       str:
         """simple docstring"""



         torch.manual_seed(0	)
         UpperCAmelCase__			=						UNetaDConditionModel(
             block_out_channels=(3_2, 6_4) ,						layers_per_block=2 ,						sample_size=3_2 ,						in_channels=4 ,						out_channels=4 ,						down_block_types=("DownBlock2D", "CrossAttnDownBlock2D") ,						up_block_types=("CrossAttnUpBlock2D", "UpBlock2D") ,						cross_attention_dim=3_2 ,						)
         UpperCAmelCase__			=						PNDMScheduler(skip_prk_steps=__UpperCAmelCase	)
         torch.manual_seed(0	)
         UpperCAmelCase__			=						AutoencoderKL(
             block_out_channels=[3_2, 6_4] ,						in_channels=3 ,						out_channels=3 ,						down_block_types=["DownEncoderBlock2D", "DownEncoderBlock2D"] ,						up_block_types=["UpDecoderBlock2D", "UpDecoderBlock2D"] ,						latent_channels=4 ,						)
         torch.manual_seed(0	)
         UpperCAmelCase__			=						CLIPVisionConfig(
             hidden_size=3_2 ,						projection_dim=3_2 ,						intermediate_size=3_7 ,						layer_norm_eps=1E-05 ,						num_attention_heads=4 ,						num_hidden_layers=5 ,						image_size=3_2 ,						patch_size=4 ,						)
         UpperCAmelCase__			=						CLIPVisionModelWithProjection(__UpperCAmelCase	)
         UpperCAmelCase__			=						CLIPImageProcessor(crop_size=3_2 ,						size=3_2	)

         UpperCAmelCase__			=						{
             "unet": unet,
             "scheduler": scheduler,
             "vae": vae,
             "image_encoder": image_encoder,
             "feature_extractor": feature_extractor,
             "safety_checker": None,
         }
         return components


   def     lowercase_   (self  :		List[str] ,						__UpperCAmelCase  :		Dict ,						__UpperCAmelCase  :		Any=0	)		->       int:
         """simple docstring"""



         UpperCAmelCase__			=						floats_tensor((1, 3, 3_2, 3_2) ,						rng=random.Random(__UpperCAmelCase	)	)
         UpperCAmelCase__			=						image.cpu().permute(0 ,						2 ,						3 ,						1	)[0]
         UpperCAmelCase__			=						Image.fromarray(np.uinta(__UpperCAmelCase	)	).convert("RGB"	).resize((3_2, 3_2)	)
         if str(__UpperCAmelCase	).startswith("mps"	):
               UpperCAmelCase__			=						torch.manual_seed(__UpperCAmelCase	)
         else:
               UpperCAmelCase__			=						torch.Generator(device=__UpperCAmelCase	).manual_seed(__UpperCAmelCase	)
         UpperCAmelCase__			=						{
             "image": image,
             "generator": generator,
             "num_inference_steps": 2,
             "guidance_scale": 6.0,
             "output_type": "numpy",
         }
         return inputs


   def     lowercase_   (self  :		Any	)		->       Any:
         """simple docstring"""



         UpperCAmelCase__			=						"cpu"  # ensure determinism for the device-dependent torch.Generator
         UpperCAmelCase__			=						self.get_dummy_components()
         UpperCAmelCase__			=						StableDiffusionImageVariationPipeline(**__UpperCAmelCase	)
         UpperCAmelCase__			=						sd_pipe.to(__UpperCAmelCase	)
         sd_pipe.set_progress_bar_config(disable=__UpperCAmelCase	)

         UpperCAmelCase__			=						self.get_dummy_inputs(__UpperCAmelCase	)
         UpperCAmelCase__			=						sd_pipe(**__UpperCAmelCase	).images
         UpperCAmelCase__			=						image[0, -3:, -3:, -1]

         assert image.shape == (1, 6_4, 6_4, 3)
         UpperCAmelCase__			=						np.array([0.5239, 0.5723, 0.4796, 0.5049, 0.5550, 0.4685, 0.5329, 0.4891, 0.4921]	)

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


   def     lowercase_   (self  :		Tuple	)		->       Tuple:
         """simple docstring"""



         UpperCAmelCase__			=						"cpu"  # ensure determinism for the device-dependent torch.Generator
         UpperCAmelCase__			=						self.get_dummy_components()
         UpperCAmelCase__			=						StableDiffusionImageVariationPipeline(**__UpperCAmelCase	)
         UpperCAmelCase__			=						sd_pipe.to(__UpperCAmelCase	)
         sd_pipe.set_progress_bar_config(disable=__UpperCAmelCase	)

         UpperCAmelCase__			=						self.get_dummy_inputs(__UpperCAmelCase	)
         UpperCAmelCase__			=						2 * [inputs["image"]]
         UpperCAmelCase__			=						sd_pipe(**__UpperCAmelCase	)

         UpperCAmelCase__			=						output.images

         UpperCAmelCase__			=						image[-1, -3:, -3:, -1]

         assert image.shape == (2, 6_4, 6_4, 3)
         UpperCAmelCase__			=						np.array([0.6892, 0.5637, 0.5836, 0.5771, 0.6254, 0.6409, 0.5580, 0.5569, 0.5289]	)

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





   def     lowercase_   (self  :		List[Any]	)		->       Union[str, Any]:
         """simple docstring"""



         super().test_inference_batch_single_identical(expected_max_diff=3E-3	)





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


   def     lowercase_   (self  :		str	)		->       Any:
         """simple docstring"""



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


   def     lowercase_   (self  :		int ,						__UpperCAmelCase  :		Tuple ,						__UpperCAmelCase  :		List[Any]="cpu" ,						__UpperCAmelCase  :		Union[str, Any]=torch.floataa ,						__UpperCAmelCase  :		List[Any]=0	)		->       Dict:
         """simple docstring"""



         UpperCAmelCase__			=						torch.Generator(device=__UpperCAmelCase	).manual_seed(__UpperCAmelCase	)
         UpperCAmelCase__			=						load_image(
             "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"
             "/stable_diffusion_imgvar/input_image_vermeer.png"	)
         UpperCAmelCase__			=						np.random.RandomState(__UpperCAmelCase	).standard_normal((1, 4, 6_4, 6_4)	)
         UpperCAmelCase__			=						torch.from_numpy(__UpperCAmelCase	).to(device=__UpperCAmelCase ,						dtype=__UpperCAmelCase	)
         UpperCAmelCase__			=						{
             "image": init_image,
             "latents": latents,
             "generator": generator,
             "num_inference_steps": 3,
             "guidance_scale": 7.5,
             "output_type": "numpy",
         }
         return inputs


   def     lowercase_   (self  :		List[str]	)		->       Dict:
         """simple docstring"""



         UpperCAmelCase__			=						StableDiffusionImageVariationPipeline.from_pretrained(
             "lambdalabs/sd-image-variations-diffusers" ,						safety_checker=__UpperCAmelCase	)
         UpperCAmelCase__			=						sd_pipe.to(__UpperCAmelCase	)
         sd_pipe.set_progress_bar_config(disable=__UpperCAmelCase	)

         UpperCAmelCase__			=						self.get_inputs(__UpperCAmelCase	)
         UpperCAmelCase__			=						sd_pipe(**__UpperCAmelCase	).images
         UpperCAmelCase__			=						image[0, -3:, -3:, -1].flatten()

         assert image.shape == (1, 5_1_2, 5_1_2, 3)
         UpperCAmelCase__			=						np.array([0.84491, 0.90789, 0.75708, 0.78734, 0.83485, 0.70099, 0.66938, 0.68727, 0.61379]	)
         assert np.abs(image_slice - expected_slice	).max() < 6E-3


   def     lowercase_   (self  :		str	)		->       Optional[Any]:
         """simple docstring"""



         UpperCAmelCase__			=						0

         def callback_fn(__UpperCAmelCase  :		int ,						__UpperCAmelCase  :		int ,						__UpperCAmelCase  :		torch.FloatTensor	) -> None:
               UpperCAmelCase__			=						True
               nonlocal number_of_steps
               number_of_steps += 1
               if step == 1:
                     UpperCAmelCase__			=						latents.detach().cpu().numpy()
                     assert latents.shape == (1, 4, 6_4, 6_4)
                     UpperCAmelCase__			=						latents[0, -3:, -3:, -1]
                     UpperCAmelCase__			=						np.array(
                         [-0.1621, 0.2837, -0.7979, -0.1221, -1.3057, 0.7681, -2.1191, 0.0464, 1.6309]	)

                     assert np.abs(latents_slice.flatten() - expected_slice	).max() < 5E-2
               elif step == 2:
                     UpperCAmelCase__			=						latents.detach().cpu().numpy()
                     assert latents.shape == (1, 4, 6_4, 6_4)
                     UpperCAmelCase__			=						latents[0, -3:, -3:, -1]
                     UpperCAmelCase__			=						np.array([0.6299, 1.7500, 1.1992, -2.1582, -1.8994, 0.7334, -0.7090, 1.0137, 1.5273]	)

                     assert np.abs(latents_slice.flatten() - expected_slice	).max() < 5E-2

         UpperCAmelCase__			=						False

         UpperCAmelCase__			=						StableDiffusionImageVariationPipeline.from_pretrained(
             "fusing/sd-image-variations-diffusers" ,						safety_checker=__UpperCAmelCase ,						torch_dtype=torch.floataa ,						)
         pipe.to(__UpperCAmelCase	)
         pipe.set_progress_bar_config(disable=__UpperCAmelCase	)
         pipe.enable_attention_slicing()

         UpperCAmelCase__			=						self.get_inputs(__UpperCAmelCase ,						dtype=torch.floataa	)
         pipe(**__UpperCAmelCase ,						callback=__UpperCAmelCase ,						callback_steps=1	)
         assert callback_fn.has_been_called
         assert number_of_steps == inputs["num_inference_steps"]





   def     lowercase_   (self  :		Optional[int]	)		->       Any:
         """simple docstring"""



         torch.cuda.empty_cache()
         torch.cuda.reset_max_memory_allocated()
         torch.cuda.reset_peak_memory_stats()

         UpperCAmelCase__			=						"fusing/sd-image-variations-diffusers"
         UpperCAmelCase__			=						StableDiffusionImageVariationPipeline.from_pretrained(
             __UpperCAmelCase ,						safety_checker=__UpperCAmelCase ,						torch_dtype=torch.floataa	)
         UpperCAmelCase__			=						pipe.to(__UpperCAmelCase	)
         pipe.set_progress_bar_config(disable=__UpperCAmelCase	)
         pipe.enable_attention_slicing(1	)
         pipe.enable_sequential_cpu_offload()

         UpperCAmelCase__			=						self.get_inputs(__UpperCAmelCase ,						dtype=torch.floataa	)
         UpperCAmelCase__			=						pipe(**__UpperCAmelCase	)

         UpperCAmelCase__			=						torch.cuda.max_memory_allocated()
         # make sure that less than 2.6 GB is allocated
         assert mem_bytes < 2.6 * 1_0**9





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


   def     lowercase_   (self  :		Union[str, Any]	)		->       Any:
         """simple docstring"""



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


   def     lowercase_   (self  :		Union[str, Any] ,						__UpperCAmelCase  :		Optional[Any] ,						__UpperCAmelCase  :		List[Any]="cpu" ,						__UpperCAmelCase  :		Tuple=torch.floataa ,						__UpperCAmelCase  :		List[Any]=0	)		->       Tuple:
         """simple docstring"""



         UpperCAmelCase__			=						torch.Generator(device=__UpperCAmelCase	).manual_seed(__UpperCAmelCase	)
         UpperCAmelCase__			=						load_image(
             "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"
             "/stable_diffusion_imgvar/input_image_vermeer.png"	)
         UpperCAmelCase__			=						np.random.RandomState(__UpperCAmelCase	).standard_normal((1, 4, 6_4, 6_4)	)
         UpperCAmelCase__			=						torch.from_numpy(__UpperCAmelCase	).to(device=__UpperCAmelCase ,						dtype=__UpperCAmelCase	)
         UpperCAmelCase__			=						{
             "image": init_image,
             "latents": latents,
             "generator": generator,
             "num_inference_steps": 5_0,
             "guidance_scale": 7.5,
             "output_type": "numpy",
         }
         return inputs


   def     lowercase_   (self  :		List[str]	)		->       Optional[int]:
         """simple docstring"""



         UpperCAmelCase__			=						StableDiffusionImageVariationPipeline.from_pretrained("fusing/sd-image-variations-diffusers"	)
         sd_pipe.to(__UpperCAmelCase	)
         sd_pipe.set_progress_bar_config(disable=__UpperCAmelCase	)

         UpperCAmelCase__			=						self.get_inputs(__UpperCAmelCase	)
         UpperCAmelCase__			=						sd_pipe(**__UpperCAmelCase	).images[0]

         UpperCAmelCase__			=						load_numpy(
             "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"
             "/stable_diffusion_imgvar/lambdalabs_variations_pndm.npy"	)
         UpperCAmelCase__			=						np.abs(expected_image - image	).max()
         assert max_diff < 1E-3





   def     lowercase_   (self  :		Dict	)		->       Dict:
         """simple docstring"""



         UpperCAmelCase__			=						StableDiffusionImageVariationPipeline.from_pretrained("fusing/sd-image-variations-diffusers"	)
         UpperCAmelCase__			=						DPMSolverMultistepScheduler.from_config(sd_pipe.scheduler.config	)
         sd_pipe.to(__UpperCAmelCase	)
         sd_pipe.set_progress_bar_config(disable=__UpperCAmelCase	)

         UpperCAmelCase__			=						self.get_inputs(__UpperCAmelCase	)
         UpperCAmelCase__			=						2_5
         UpperCAmelCase__			=						sd_pipe(**__UpperCAmelCase	).images[0]

         UpperCAmelCase__			=						load_numpy(
             "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"
             "/stable_diffusion_imgvar/lambdalabs_variations_dpm_multi.npy"	)
         UpperCAmelCase__			=						np.abs(expected_image - image	).max()
         assert max_diff < 1E-3




UpperCamelCase__       							=      [
    [0, 0, 1, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0],
    [0, 0, 0, 0, 0, 0, 0, 1, 1, 1, 0, 0, 0],
    [0, 1, 1, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0],
    [0, 1, 0, 0, 1, 1, 0, 0, 1, 0, 1, 0, 0],
    [0, 1, 0, 0, 1, 1, 0, 0, 1, 1, 1, 0, 0],
    [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0],
    [0, 0, 0, 0, 0, 0, 0, 1, 1, 1, 0, 0, 0],
    [0, 0, 0, 0, 0, 0, 0, 1, 1, 0, 0, 0, 0],
]




def      lowerCAmelCase_				(  __A,						__A,						__A,						__A							)  ->       bool:







      '''simple docstring'''


      return 0 <= row < rows and 0 <= col < cols




def      lowerCAmelCase_				(  __A,						__A,						__A,						__A							)  ->       int:







      '''simple docstring'''


      UpperCAmelCase__			=						len(__A							)
      UpperCAmelCase__			=						len(mat[0]							)
      if is_safe(__A,						__A,						__A,						__A							) and (row, col) not in seen and mat[row][col] == 1:
            seen.add((row, col)							)
            return (
                1
                + depth_first_search(row + 1,						__A,						__A,						__A							)
                + depth_first_search(row - 1,						__A,						__A,						__A							)
                + depth_first_search(__A,						col + 1,						__A,						__A							)
                + depth_first_search(__A,						col - 1,						__A,						__A							)
            )
      else:
            return 0




def      lowerCAmelCase_				(  __A							)  ->       int:







      '''simple docstring'''


      UpperCAmelCase__			=						set()

      UpperCAmelCase__			=						0
      for row, line in enumerate(__A							):
            for col, item in enumerate(__A							):
                  if item == 1 and (row, col) not in seen:
                        # Maximizing the area
                        UpperCAmelCase__			=						max(__A,						depth_first_search(__A,						__A,						__A,						__A							)							)
      return max_area


if __name__ == "__main__":
      import doctest

      print(find_max_area(matrix))  # Output -> 6
      doctest.testmod()



