"""simple docstring"""




import gc
import random
import unittest

import numpy as np
import torch
from transformers import (
    CLIPImageProcessor,
    CLIPTextConfig,
    CLIPTextModel,
    CLIPTokenizer,
    CLIPVisionConfig,
    CLIPVisionModelWithProjection,
)

from diffusers import AutoencoderKL, DDIMScheduler, DDPMScheduler, StableUnCLIPImgaImgPipeline, UNetaDConditionModel
from diffusers.pipelines.pipeline_utils import DiffusionPipeline
from diffusers.pipelines.stable_diffusion.stable_unclip_image_normalizer import StableUnCLIPImageNormalizer
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    floats_tensor,
    load_image,
    load_numpy,
    require_torch_gpu,
    skip_mps,
    slow,
    torch_device,
)

from ..pipeline_params import TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS, TEXT_GUIDED_IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
    assert_mean_pixel_difference,
)


enable_full_determinism()
class 						__A							( SCREAMING_SNAKE_CASE_    ,	SCREAMING_SNAKE_CASE_    ,	SCREAMING_SNAKE_CASE_    ,	unittest.TestCase		):
     _UpperCamelCase							:			Dict							=		StableUnCLIPImgaImgPipeline
     _UpperCamelCase							:			Optional[int]							=		TEXT_GUIDED_IMAGE_VARIATION_PARAMS
     _UpperCamelCase							:			List[Any]							=		TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
     _UpperCamelCase							:			Tuple							=		frozenset(
         []		)  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
     _UpperCamelCase							:			List[str]							=		frozenset([]		)






     def     __A  (		self  ):
       _lowerCAmelCase						:       Any  			=     32
       _lowerCAmelCase						:       str  			=     embedder_hidden_size

       # image encoding components

       _lowerCAmelCase						:       int  			=     CLIPImageProcessor(crop_size=32				,     size=32  )

       torch.manual_seed(0  )
       _lowerCAmelCase						:       Optional[Any]  			=     CLIPVisionModelWithProjection(
           CLIPVisionConfig(
               hidden_size=a__				,     projection_dim=a__				,     num_hidden_layers=5				,     num_attention_heads=4				,     image_size=32				,     intermediate_size=37				,     patch_size=1				,     )  )

       # regular denoising components

       torch.manual_seed(0  )
       _lowerCAmelCase						:       Tuple  			=     StableUnCLIPImageNormalizer(embedding_dim=a__  )
       _lowerCAmelCase						:       str  			=     DDPMScheduler(beta_schedule="""squaredcos_cap_v2"""  )

       torch.manual_seed(0  )
       _lowerCAmelCase						:       List[str]  			=     CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""  )

       torch.manual_seed(0  )
       _lowerCAmelCase						:       Union[str, Any]  			=     CLIPTextModel(
           CLIPTextConfig(
               bos_token_id=0				,     eos_token_id=2				,     hidden_size=a__				,     projection_dim=32				,     intermediate_size=37				,     layer_norm_eps=1e-05				,     num_attention_heads=4				,     num_hidden_layers=5				,     pad_token_id=1				,     vocab_size=1000				,     )  )

       torch.manual_seed(0  )
       _lowerCAmelCase						:       str  			=     UNetaDConditionModel(
           sample_size=32				,     in_channels=4				,     out_channels=4				,     down_block_types=("""CrossAttnDownBlock2D""", """DownBlock2D""")				,     up_block_types=("""UpBlock2D""", """CrossAttnUpBlock2D""")				,     block_out_channels=(32, 64)				,     attention_head_dim=(2, 4)				,     class_embed_type="""projection"""				,     projection_class_embeddings_input_dim=embedder_projection_dim * 2				,     cross_attention_dim=a__				,     layers_per_block=1				,     upcast_attention=a__				,     use_linear_projection=a__				,     )

       torch.manual_seed(0  )
       _lowerCAmelCase						:       Any  			=     DDIMScheduler(
           beta_schedule="""scaled_linear"""				,     beta_start=0.0_0_0_8_5				,     beta_end=0.0_1_2				,     prediction_type="""v_prediction"""				,     set_alpha_to_one=a__				,     steps_offset=1				,     )

       torch.manual_seed(0  )
       _lowerCAmelCase						:       int  			=     AutoencoderKL()

       _lowerCAmelCase						:       Any  			=     {
           # image encoding components
           """feature_extractor""": feature_extractor,
           """image_encoder""": image_encoder.eval(),
           # image noising components
           """image_normalizer""": image_normalizer.eval(),
           """image_noising_scheduler""": image_noising_scheduler,
           # regular denoising components
           """tokenizer""": tokenizer,
           """text_encoder""": text_encoder.eval(),
           """unet""": unet.eval(),
           """scheduler""": scheduler,
           """vae""": vae.eval(),
       }

       return components






     def     __A  (		self				,     a__				,     a__=0				,     a__=True  ):
       if str(a__  ).startswith("""mps"""  ):
         _lowerCAmelCase						:       Optional[int]  			=     torch.manual_seed(a__  )
       else:
         _lowerCAmelCase						:       Tuple  			=     torch.Generator(device=a__  ).manual_seed(a__  )

       _lowerCAmelCase						:       List[str]  			=     floats_tensor((1, 3, 32, 32)				,     rng=random.Random(a__  )  ).to(a__  )

       if pil_image:
         _lowerCAmelCase						:       Dict  			=     input_image * 0.5 + 0.5
         _lowerCAmelCase						:       Any  			=     input_image.clamp(0				,     1  )
         _lowerCAmelCase						:       List[Any]  			=     input_image.cpu().permute(0				,     2				,     3				,     1  ).float().numpy()
         _lowerCAmelCase						:       Dict  			=     DiffusionPipeline.numpy_to_pil(a__  )[0]

       return {
           "prompt": "An anime racoon running a marathon",
           "image": input_image,
           "generator": generator,
           "num_inference_steps": 2,
           "output_type": "np",
       }






     @skip_mps
     def     __A  (		self  ):
       _lowerCAmelCase						:       Optional[Any]  			=     """cpu"""  # ensure determinism for the device-dependent torch.Generator
       _lowerCAmelCase						:       Dict  			=     self.get_dummy_components()
       _lowerCAmelCase						:       Tuple  			=     StableUnCLIPImgaImgPipeline(**a__  )
       _lowerCAmelCase						:       List[str]  			=     sd_pipe.to(a__  )
       sd_pipe.set_progress_bar_config(disable=a__  )

       _lowerCAmelCase						:       str  			=     self.get_dummy_inputs(a__  )
       inputs.update({"""image_embeds""": None}  )
       _lowerCAmelCase						:       Any  			=     sd_pipe(**a__  ).images
       _lowerCAmelCase						:       Tuple  			=     image[0, -3:, -3:, -1]

       assert image.shape == (1, 32, 32, 3)
       _lowerCAmelCase						:       List[str]  			=     np.array([0.3_8_7_2, 0.7_2_2_4, 0.5_6_0_1, 0.4_7_4_1, 0.6_8_7_2, 0.5_8_1_4, 0.4_6_3_6, 0.3_8_6_7, 0.5_0_7_8]  )

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






     def     __A  (		self  ):
       _lowerCAmelCase						:       str  			=     torch_device in ["""cpu""", """mps"""]

       self._test_attention_slicing_forward_pass(test_max_difference=a__  )






     def     __A  (		self  ):
       _lowerCAmelCase						:       str  			=     torch_device in ["""cpu""", """mps"""]

       self._test_inference_batch_single_identical(test_max_difference=a__  )






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







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






     def     __A  (		self  ):
       # clean up the VRAM after each test
       super().tearDown()
       gc.collect()
       torch.cuda.empty_cache()






     def     __A  (		self  ):
       _lowerCAmelCase						:       Union[str, Any]  			=     load_image(
           """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png"""  )

       _lowerCAmelCase						:       Optional[int]  			=     load_numpy(
           """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_img2img_anime_turtle_fp16.npy"""  )

       _lowerCAmelCase						:       Union[str, Any]  			=     StableUnCLIPImgaImgPipeline.from_pretrained(
           """fusing/stable-unclip-2-1-l-img2img"""				,     torch_dtype=torch.floataa  )
       pipe.to(a__  )
       pipe.set_progress_bar_config(disable=a__  )
       # stable unclip will oom when integration tests are run on a V100,
       # so turn on memory savings
       pipe.enable_attention_slicing()
       pipe.enable_sequential_cpu_offload()

       _lowerCAmelCase						:       List[str]  			=     torch.Generator(device="""cpu"""  ).manual_seed(0  )
       _lowerCAmelCase						:       List[Any]  			=     pipe(a__				,     """anime turle"""				,     generator=a__				,     output_type="""np"""  )

       _lowerCAmelCase						:       str  			=     output.images[0]

       assert image.shape == (768, 768, 3)

       assert_mean_pixel_difference(a__				,     a__  )






     def     __A  (		self  ):
       _lowerCAmelCase						:       Any  			=     load_image(
           """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png"""  )

       _lowerCAmelCase						:       Any  			=     load_numpy(
           """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_h_img2img_anime_turtle_fp16.npy"""  )

       _lowerCAmelCase						:       Tuple  			=     StableUnCLIPImgaImgPipeline.from_pretrained(
           """fusing/stable-unclip-2-1-h-img2img"""				,     torch_dtype=torch.floataa  )
       pipe.to(a__  )
       pipe.set_progress_bar_config(disable=a__  )
       # stable unclip will oom when integration tests are run on a V100,
       # so turn on memory savings
       pipe.enable_attention_slicing()
       pipe.enable_sequential_cpu_offload()

       _lowerCAmelCase						:       str  			=     torch.Generator(device="""cpu"""  ).manual_seed(0  )
       _lowerCAmelCase						:       Tuple  			=     pipe(a__				,     """anime turle"""				,     generator=a__				,     output_type="""np"""  )

       _lowerCAmelCase						:       List[str]  			=     output.images[0]

       assert image.shape == (768, 768, 3)

       assert_mean_pixel_difference(a__				,     a__  )






     def     __A  (		self  ):
       _lowerCAmelCase						:       int  			=     load_image(
           """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png"""  )

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

       _lowerCAmelCase						:       Any  			=     StableUnCLIPImgaImgPipeline.from_pretrained(
           """fusing/stable-unclip-2-1-h-img2img"""				,     torch_dtype=torch.floataa  )
       _lowerCAmelCase						:       List[str]  			=     pipe.to(a__  )
       pipe.set_progress_bar_config(disable=a__  )
       pipe.enable_attention_slicing()
       pipe.enable_sequential_cpu_offload()

       _lowerCAmelCase						:       List[str]  			=     pipe(
           a__				,     """anime turtle"""				,     num_inference_steps=2				,     output_type="""np"""				,     )

       _lowerCAmelCase						:       int  			=     torch.cuda.max_memory_allocated()
       # make sure that less than 7 GB is allocated
       assert mem_bytes < 7 * 10**9





"""simple docstring"""






from collections.abc import Sequence



def SCREAMING_SNAKE_CASE__			(   __UpperCAmelCase       ,    __UpperCAmelCase = False )  ->  float:
   if not arr:
      return 0

   lowercase__:   Any					=      0 if allow_empty_subarrays else float('''-inf''' )
   lowercase__:   Union[str, Any]					=      0.0
   for num in arr:
      lowercase__:   List[str]					=      max(0 if allow_empty_subarrays else num       ,    curr_sum + num )
      lowercase__:   int					=      max(__UpperCAmelCase       ,    __UpperCAmelCase )

   return max_sum


if __name__ == "__main__":
       from doctest import testmod

       testmod()

       __A			   = [-2, 1, -3, 4, -1, 2, 1, -5, 4]
       print(f'''{max_subarray_sum(nums) = }''')
