




'''simple docstring'''



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     lowerCAmelCase   ( UpperCamelCase__				,   UpperCamelCase__				,   UpperCamelCase__				,   unittest.TestCase):
       __lowercase  :      Optional[Any]        =			StableDiffusionImageVariationPipeline
       __lowercase  :      Optional[int]        =			IMAGE_VARIATION_PARAMS
       __lowercase  :      Optional[int]        =			IMAGE_VARIATION_BATCH_PARAMS
       __lowercase  :      Optional[int]        =			frozenset([])
       # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
       __lowercase  :      str        =			frozenset([])


       def       lowerCAmelCase      (   self  )      ->  str:



            '''simple docstring'''




            torch.manual_seed(0  )
            __snake_case    					=  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    ,  )
            __snake_case    					=  PNDMScheduler(skip_prk_steps=lowerCAmelCase__  )
            torch.manual_seed(0  )
            __snake_case    					=  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    ,  )
            torch.manual_seed(0  )
            __snake_case    					=  CLIPVisionConfig(
                hidden_size=32    ,  projection_dim=32    ,  intermediate_size=37    ,  layer_norm_eps=1E-05    ,  num_attention_heads=4    ,  num_hidden_layers=5    ,  image_size=32    ,  patch_size=4    ,  )
            __snake_case    					=  CLIPVisionModelWithProjection(lowerCAmelCase__  )
            __snake_case    					=  CLIPImageProcessor(crop_size=32    ,  size=32  )

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


       def       lowerCAmelCase      (   self    ,  __SCREAMING_SNAKE_CASE    ,  __SCREAMING_SNAKE_CASE=0  )      ->  List[str]:



            '''simple docstring'''




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


       def       lowerCAmelCase      (   self  )      ->  Optional[int]:



            '''simple docstring'''




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

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

            assert image.shape == (1, 64, 64, 3)
            __snake_case    					=  np.array([0.5_239, 0.5_723, 0.4_796, 0.5_049, 0.5_550, 0.4_685, 0.5_329, 0.4_891, 0.4_921]  )

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


       def       lowerCAmelCase      (   self  )      ->  List[Any]:



            '''simple docstring'''




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

            __snake_case    					=  self.get_dummy_inputs(lowerCAmelCase__  )
            __snake_case    					=  2 * [inputs["image"]]
            __snake_case    					=  sd_pipe(**lowerCAmelCase__  )

            __snake_case    					=  output.images

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

            assert image.shape == (2, 64, 64, 3)
            __snake_case    					=  np.array([0.6_892, 0.5_637, 0.5_836, 0.5_771, 0.6_254, 0.6_409, 0.5_580, 0.5_569, 0.5_289]  )

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




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



            '''simple docstring'''




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





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


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



            '''simple docstring'''




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


       def       lowerCAmelCase      (   self    ,  __SCREAMING_SNAKE_CASE    ,  __SCREAMING_SNAKE_CASE="cpu"    ,  __SCREAMING_SNAKE_CASE=torch.floataa    ,  __SCREAMING_SNAKE_CASE=0  )      ->  int:



            '''simple docstring'''




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


       def       lowerCAmelCase      (   self  )      ->  List[Any]:



            '''simple docstring'''




            __snake_case    					=  StableDiffusionImageVariationPipeline.from_pretrained(
                '''lambdalabs/sd-image-variations-diffusers'''    ,  safety_checker=lowerCAmelCase__  )
            __snake_case    					=  sd_pipe.to(lowerCAmelCase__  )
            sd_pipe.set_progress_bar_config(disable=lowerCAmelCase__  )

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

            assert image.shape == (1, 512, 512, 3)
            __snake_case    					=  np.array([0.84_491, 0.90_789, 0.75_708, 0.78_734, 0.83_485, 0.70_099, 0.66_938, 0.68_727, 0.61_379]  )
            assert np.abs(image_slice - expected_slice  ).max() < 6E-3


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



            '''simple docstring'''




            __snake_case    					=  0

            def callback_fn(__SCREAMING_SNAKE_CASE    ,  __SCREAMING_SNAKE_CASE    ,  __SCREAMING_SNAKE_CASE  ) -> None:
                 __snake_case    					=  True
                 nonlocal number_of_steps
                 number_of_steps += 1
                 if step == 1:
                      __snake_case    					=  latents.detach().cpu().numpy()
                      assert latents.shape == (1, 4, 64, 64)
                      __snake_case    					=  latents[0, -3:, -3:, -1]
                      __snake_case    					=  np.array(
                          [-0.1_621, 0.2_837, -0.7_979, -0.1_221, -1.3_057, 0.7_681, -2.1_191, 0.0_464, 1.6_309]  )

                      assert np.abs(latents_slice.flatten() - expected_slice  ).max() < 5E-2
                 elif step == 2:
                      __snake_case    					=  latents.detach().cpu().numpy()
                      assert latents.shape == (1, 4, 64, 64)
                      __snake_case    					=  latents[0, -3:, -3:, -1]
                      __snake_case    					=  np.array([0.6_299, 1.7_500, 1.1_992, -2.1_582, -1.8_994, 0.7_334, -0.7_090, 1.0_137, 1.5_273]  )

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

            __snake_case    					=  False

            __snake_case    					=  StableDiffusionImageVariationPipeline.from_pretrained(
                '''fusing/sd-image-variations-diffusers'''    ,  safety_checker=lowerCAmelCase__    ,  torch_dtype=torch.floataa    ,  )
            pipe.to(lowerCAmelCase__  )
            pipe.set_progress_bar_config(disable=lowerCAmelCase__  )
            pipe.enable_attention_slicing()

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




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



            '''simple docstring'''




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

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

            __snake_case    					=  self.get_inputs(lowerCAmelCase__    ,  dtype=torch.floataa  )
            __snake_case    					=  pipe(**lowerCAmelCase__  )

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





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


       def       lowerCAmelCase      (   self  )      ->  Dict:



            '''simple docstring'''




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


       def       lowerCAmelCase      (   self    ,  __SCREAMING_SNAKE_CASE    ,  __SCREAMING_SNAKE_CASE="cpu"    ,  __SCREAMING_SNAKE_CASE=torch.floataa    ,  __SCREAMING_SNAKE_CASE=0  )      ->  Dict:



            '''simple docstring'''




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


       def       lowerCAmelCase      (   self  )      ->  Any:



            '''simple docstring'''




            __snake_case    					=  StableDiffusionImageVariationPipeline.from_pretrained('''fusing/sd-image-variations-diffusers'''  )
            sd_pipe.to(lowerCAmelCase__  )
            sd_pipe.set_progress_bar_config(disable=lowerCAmelCase__  )

            __snake_case    					=  self.get_inputs(lowerCAmelCase__  )
            __snake_case    					=  sd_pipe(**lowerCAmelCase__  ).images[0]

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




       def       lowerCAmelCase      (   self  )      ->  Optional[int]:



            '''simple docstring'''




            __snake_case    					=  StableDiffusionImageVariationPipeline.from_pretrained('''fusing/sd-image-variations-diffusers'''  )
            __snake_case    					=  DPMSolverMultistepScheduler.from_config(sd_pipe.scheduler.config  )
            sd_pipe.to(lowerCAmelCase__  )
            sd_pipe.set_progress_bar_config(disable=lowerCAmelCase__  )

            __snake_case    					=  self.get_inputs(lowerCAmelCase__  )
            __snake_case    					=  25
            __snake_case    					=  sd_pipe(**lowerCAmelCase__  ).images[0]

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












'''simple docstring'''



# This Python program implements an optimal binary search tree (abbreviated BST)
# building dynamic programming algorithm that delivers O(n^2) performance.
#
# The goal of the optimal BST problem is to build a low-cost BST for a
# given set of nodes, each with its own key and frequency. The frequency
# of the node is defined as how many time the node is being searched.
# The search cost of binary search tree is given by this formula:
#
# cost(1, n) = sum{i = 1 to n}((depth(node_i) + 1) * node_i_freq)
#
# where n is number of nodes in the BST. The characteristic of low-cost
# BSTs is having a faster overall search time than other implementations.
# The reason for their fast search time is that the nodes with high
# frequencies will be placed near the root of the tree while the nodes
# with low frequencies will be placed near the leaves of the tree thus
# reducing search time in the most frequent instances.
import sys
from random import randint





class     lowerCAmelCase   :


       def __init__(   self    ,  __SCREAMING_SNAKE_CASE    ,  __SCREAMING_SNAKE_CASE  )      ->  Optional[int]:



            '''simple docstring'''




            __snake_case    					=  key
            __snake_case    					=  freq




       def __str__(   self  )      ->  List[Any]:



            '''simple docstring'''




            return F'''Node(key={self.key}, freq={self.freq})'''







def 			_UpperCamelCase (_lowerCamelCase							:     str   ,	_lowerCamelCase							:     List[str]   ,	_lowerCamelCase							:     Union[str, Any]   ,	_lowerCamelCase							:     int   ,	_lowerCamelCase							:     List[Any]   ,	_lowerCamelCase							:     Tuple	)->   Any:




     '''simple docstring'''





     if i > j or i < 0 or j > len(_lowerCamelCase	) - 1:
          return

     __snake_case    					=  root[i][j]
     if parent == -1:  # root does not have a parent
          print(f'''{key[node]} is the root of the binary search tree.'''	)
     elif is_left:
          print(f'''{key[node]} is the left child of key {parent}.'''	)
     else:
          print(f'''{key[node]} is the right child of key {parent}.'''	)

     print_binary_search_tree(_lowerCamelCase   ,	_lowerCamelCase   ,	_lowerCamelCase   ,	node - 1   ,	key[node]   ,	_lowerCamelCase	)
     print_binary_search_tree(_lowerCamelCase   ,	_lowerCamelCase   ,	node + 1   ,	_lowerCamelCase   ,	key[node]   ,	_lowerCamelCase	)







def 			_UpperCamelCase (_lowerCamelCase							:     List[Any]	)->   Union[str, Any]:




     '''simple docstring'''





     nodes.sort(key=lambda _lowerCamelCase	: node.key	)

     __snake_case    					=  len(_lowerCamelCase	)

     __snake_case    					=  [nodes[i].key for i in range(_lowerCamelCase	)]
     __snake_case    					=  [nodes[i].freq for i in range(_lowerCamelCase	)]

     # This 2D array stores the overall tree cost (which's as minimized as possible);
     # for a single key, cost is equal to frequency of the key.
     __snake_case    					=  [[freqs[i] if i == j else 0 for j in range(_lowerCamelCase	)] for i in range(_lowerCamelCase	)]
     # sum[i][j] stores the sum of key frequencies between i and j inclusive in nodes
     # array
     __snake_case    					=  [[freqs[i] if i == j else 0 for j in range(_lowerCamelCase	)] for i in range(_lowerCamelCase	)]
     # stores tree roots that will be used later for constructing binary search tree
     __snake_case    					=  [[i if i == j else 0 for j in range(_lowerCamelCase	)] for i in range(_lowerCamelCase	)]

     for interval_length in range(2   ,	n + 1	):
          for i in range(n - interval_length + 1	):
               __snake_case    					=  i + interval_length - 1

               __snake_case    					=  sys.maxsize  # set the value to "infinity"
               __snake_case    					=  total[i][j - 1] + freqs[j]

               # Apply Knuth's optimization
               # Loop without optimization: for r in range(i, j + 1):
               for r in range(root[i][j - 1]   ,	root[i + 1][j] + 1	):  # r is a temporal root
                    __snake_case    					=  dp[i][r - 1] if r != i else 0  # optimal cost for left subtree
                    __snake_case    					=  dp[r + 1][j] if r != j else 0  # optimal cost for right subtree
                    __snake_case    					=  left + total[i][j] + right

                    if dp[i][j] > cost:
                         __snake_case    					=  cost
                         __snake_case    					=  r

     print('''Binary search tree nodes:'''	)
     for node in nodes:
          print(_lowerCamelCase	)

     print(f'''\nThe cost of optimal BST for given tree nodes is {dp[0][n - 1]}.'''	)
     print_binary_search_tree(_lowerCamelCase   ,	_lowerCamelCase   ,	0   ,	n - 1   ,	-1   ,	_lowerCamelCase	)







def 			_UpperCamelCase ()->   Union[str, Any]:




     '''simple docstring'''





     __snake_case    					=  [Node(_lowerCamelCase   ,	randint(1   ,	50	)	) for i in range(10   ,	0   ,	-1	)]
     find_optimal_binary_search_tree(_lowerCamelCase	)


if __name__ == "__main__":
     main()






