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 		_lowerCamelCase		(   UpperCamelCase_    ,     UpperCamelCase_    ,     UpperCamelCase_    ,     unittest.TestCase    ):
      __a  =      StableUnCLIPImgaImgPipeline
      __a  =      TEXT_GUIDED_IMAGE_VARIATION_PARAMS
      __a  =      TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
      __a  =      frozenset(
          []    )  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
      __a  =      frozenset([]    )
      def 						UpperCamelCase_				(  self   )       ->					Dict:
       SCREAMING_SNAKE_CASE__:   Union[str, Any]=      32
       SCREAMING_SNAKE_CASE__:   Optional[int]=      embedder_hidden_size

       # image encoding components

       SCREAMING_SNAKE_CASE__:   Any=      CLIPImageProcessor(crop_size=32 , size=32   )

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

       # regular denoising components

       torch.manual_seed(0   )
       SCREAMING_SNAKE_CASE__:   Any=      StableUnCLIPImageNormalizer(embedding_dim=lowerCAmelCase   )
       SCREAMING_SNAKE_CASE__:   Union[str, Any]=      DDPMScheduler(beta_schedule='''squaredcos_cap_v2'''   )

       torch.manual_seed(0   )
       SCREAMING_SNAKE_CASE__:   Any=      CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''   )

       torch.manual_seed(0   )
       SCREAMING_SNAKE_CASE__:   Optional[Any]=      CLIPTextModel(
           CLIPTextConfig(
               bos_token_id=0 , eos_token_id=2 , hidden_size=lowerCAmelCase , 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   )
       SCREAMING_SNAKE_CASE__:   Optional[Any]=      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=lowerCAmelCase , layers_per_block=1 , upcast_attention=lowerCAmelCase , use_linear_projection=lowerCAmelCase , )

       torch.manual_seed(0   )
       SCREAMING_SNAKE_CASE__:   Tuple=      DDIMScheduler(
           beta_schedule='''scaled_linear''' , beta_start=0.00085 , beta_end=0.012 , prediction_type='''v_prediction''' , set_alpha_to_one=lowerCAmelCase , steps_offset=1 , )

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

       SCREAMING_SNAKE_CASE__:   List[str]=      {
           # 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 						UpperCamelCase_				(  self , lowerCAmelCase , lowerCAmelCase=0 , lowerCAmelCase=True   )       ->					Tuple:
       if str(lowerCAmelCase   ).startswith('''mps'''   ):
        SCREAMING_SNAKE_CASE__:   Tuple=      torch.manual_seed(lowerCAmelCase   )
       else:
        SCREAMING_SNAKE_CASE__:   int=      torch.Generator(device=lowerCAmelCase   ).manual_seed(lowerCAmelCase   )

       SCREAMING_SNAKE_CASE__:   int=      floats_tensor((1, 3, 32, 32) , rng=random.Random(lowerCAmelCase   )   ).to(lowerCAmelCase   )

       if pil_image:
        SCREAMING_SNAKE_CASE__:   Union[str, Any]=      input_image * 0.5 + 0.5
        SCREAMING_SNAKE_CASE__:   Tuple=      input_image.clamp(0 , 1   )
        SCREAMING_SNAKE_CASE__:   Optional[int]=      input_image.cpu().permute(0 , 2 , 3 , 1   ).float().numpy()
        SCREAMING_SNAKE_CASE__:   Optional[int]=      DiffusionPipeline.numpy_to_pil(lowerCAmelCase   )[0]

       return {
           "prompt": "An anime racoon running a marathon",
           "image": input_image,
           "generator": generator,
           "num_inference_steps": 2,
           "output_type": "np",
       }
      @skip_mps
      def 						UpperCamelCase_				(  self   )       ->					Optional[Any]:
       SCREAMING_SNAKE_CASE__:   str=      '''cpu'''  # ensure determinism for the device-dependent torch.Generator
       SCREAMING_SNAKE_CASE__:   Optional[int]=      self.get_dummy_components()
       SCREAMING_SNAKE_CASE__:   Tuple=      StableUnCLIPImgaImgPipeline(**lowerCAmelCase   )
       SCREAMING_SNAKE_CASE__:   int=      sd_pipe.to(lowerCAmelCase   )
       sd_pipe.set_progress_bar_config(disable=lowerCAmelCase   )

       SCREAMING_SNAKE_CASE__:   int=      self.get_dummy_inputs(lowerCAmelCase   )
       inputs.update({'''image_embeds''': None}   )
       SCREAMING_SNAKE_CASE__:   List[Any]=      sd_pipe(**lowerCAmelCase   ).images
       SCREAMING_SNAKE_CASE__:   Dict=      image[0, -3:, -3:, -1]

       assert image.shape == (1, 32, 32, 3)
       SCREAMING_SNAKE_CASE__:   Optional[int]=      np.array([0.3872, 0.7224, 0.5601, 0.4741, 0.6872, 0.5814, 0.4636, 0.3867, 0.5078]   )

       assert np.abs(image_slice.flatten() - expected_slice   ).max() < 1e-3
      def 						UpperCamelCase_				(  self   )       ->					List[str]:
       SCREAMING_SNAKE_CASE__:   List[Any]=      torch_device in ['''cpu''', '''mps''']

       self._test_attention_slicing_forward_pass(test_max_difference=lowerCAmelCase   )
      def 						UpperCamelCase_				(  self   )       ->					Tuple:
       SCREAMING_SNAKE_CASE__:   Optional[Any]=      torch_device in ['''cpu''', '''mps''']

       self._test_inference_batch_single_identical(test_max_difference=lowerCAmelCase   )



      @unittest.skipIf(
          torch_device != '''cuda''' or not is_xformers_available() , reason='''XFormers attention is only available with CUDA and `xformers` installed''' , )
      def 						UpperCamelCase_				(  self   )       ->					Any:
       self._test_xformers_attention_forwardGenerator_pass(test_max_difference=lowerCAmelCase   )






@slow
@require_torch_gpu
class 		_lowerCamelCase		(   unittest.TestCase    ):
      def 						UpperCamelCase_				(  self   )       ->					Optional[Any]:
       # clean up the VRAM after each test
       super().tearDown()
       gc.collect()
       torch.cuda.empty_cache()
      def 						UpperCamelCase_				(  self   )       ->					str:
       SCREAMING_SNAKE_CASE__:   Tuple=      load_image(
           '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png'''   )

       SCREAMING_SNAKE_CASE__:   List[Any]=      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'''   )

       SCREAMING_SNAKE_CASE__:   List[str]=      StableUnCLIPImgaImgPipeline.from_pretrained(
           '''fusing/stable-unclip-2-1-l-img2img''' , torch_dtype=torch.floataa   )
       pipe.to(lowerCAmelCase   )
       pipe.set_progress_bar_config(disable=lowerCAmelCase   )
       # 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()

       SCREAMING_SNAKE_CASE__:   List[Any]=      torch.Generator(device='''cpu'''   ).manual_seed(0   )
       SCREAMING_SNAKE_CASE__:   Dict=      pipe(lowerCAmelCase , '''anime turle''' , generator=lowerCAmelCase , output_type='''np'''   )

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

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

       assert_mean_pixel_difference(lowerCAmelCase , lowerCAmelCase   )
      def 						UpperCamelCase_				(  self   )       ->					Optional[int]:
       SCREAMING_SNAKE_CASE__:   Dict=      load_image(
           '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png'''   )

       SCREAMING_SNAKE_CASE__:   List[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'''   )

       SCREAMING_SNAKE_CASE__:   Any=      StableUnCLIPImgaImgPipeline.from_pretrained(
           '''fusing/stable-unclip-2-1-h-img2img''' , torch_dtype=torch.floataa   )
       pipe.to(lowerCAmelCase   )
       pipe.set_progress_bar_config(disable=lowerCAmelCase   )
       # 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()

       SCREAMING_SNAKE_CASE__:   List[str]=      torch.Generator(device='''cpu'''   ).manual_seed(0   )
       SCREAMING_SNAKE_CASE__:   Any=      pipe(lowerCAmelCase , '''anime turle''' , generator=lowerCAmelCase , output_type='''np'''   )

       SCREAMING_SNAKE_CASE__:   str=      output.images[0]

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

       assert_mean_pixel_difference(lowerCAmelCase , lowerCAmelCase   )



      def 						UpperCamelCase_				(  self   )       ->					Any:
       SCREAMING_SNAKE_CASE__:   str=      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()

       SCREAMING_SNAKE_CASE__:   Tuple=      StableUnCLIPImgaImgPipeline.from_pretrained(
           '''fusing/stable-unclip-2-1-h-img2img''' , torch_dtype=torch.floataa   )
       SCREAMING_SNAKE_CASE__:   Any=      pipe.to(lowerCAmelCase   )
       pipe.set_progress_bar_config(disable=lowerCAmelCase   )
       pipe.enable_attention_slicing()
       pipe.enable_sequential_cpu_offload()

       SCREAMING_SNAKE_CASE__:   List[str]=      pipe(
           lowerCAmelCase , '''anime turtle''' , num_inference_steps=2 , output_type='''np''' , )

       SCREAMING_SNAKE_CASE__:   List[Any]=      torch.cuda.max_memory_allocated()
       # make sure that less than 7 GB is allocated
       assert mem_bytes < 7 * 10**9


import json
import os
from collections import Counter

import torch
import torchvision
import torchvision.transforms as transforms
from PIL import Image
from torch import nn
from torch.utils.data import Dataset


lowercase_     :   List[str] 				=				{1: (1, 1), 2: (2, 1), 3: (3, 1), 4: (2, 2), 5: (5, 1), 6: (3, 2), 7: (7, 1), 8: (4, 2), 9: (3, 3)}






class 		_lowerCamelCase		(   nn.Module    ):
    def __init__(  self , lowerCAmelCase   )       ->					str:
     super().__init__()
     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      torchvision.models.resnetaaa(pretrained=lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   Optional[Any]=      list(model.children()   )[:-2]
     SCREAMING_SNAKE_CASE__:   Tuple=      nn.Sequential(*lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   Optional[Any]=      nn.AdaptiveAvgPoolad(POOLING_BREAKDOWN[args.num_image_embeds]   )



    def 						UpperCamelCase_				(  self , lowerCAmelCase   )       ->					List[str]:
     # Bx3x224x224 -> Bx2048x7x7 -> Bx2048xN -> BxNx2048
     SCREAMING_SNAKE_CASE__:   str=      self.pool(self.model(lowerCAmelCase   )   )
     SCREAMING_SNAKE_CASE__:   str=      torch.flatten(lowerCAmelCase , start_dim=2   )
     SCREAMING_SNAKE_CASE__:   Optional[Any]=      out.transpose(1 , 2   ).contiguous()
     return out  # BxNx2048






class 		_lowerCamelCase		(   UpperCamelCase_    ):
    def __init__(  self , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase   )       ->					Any:
     SCREAMING_SNAKE_CASE__:   Optional[int]=      [json.loads(lowerCAmelCase   ) for l in open(lowerCAmelCase   )]
     SCREAMING_SNAKE_CASE__:   List[str]=      os.path.dirname(lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   int=      tokenizer
     SCREAMING_SNAKE_CASE__:   Any=      labels
     SCREAMING_SNAKE_CASE__:   List[Any]=      len(lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   Any=      max_seq_length

     SCREAMING_SNAKE_CASE__:   Optional[Any]=      transforms
    def __len__(  self   )       ->					Optional[Any]:
     return len(self.data   )
    def __getitem__(  self , lowerCAmelCase   )       ->					Tuple:
     SCREAMING_SNAKE_CASE__:   Optional[Any]=      torch.LongTensor(self.tokenizer.encode(self.data[index]['''text'''] , add_special_tokens=lowerCAmelCase   )   )
     SCREAMING_SNAKE_CASE__,	SCREAMING_SNAKE_CASE__,	SCREAMING_SNAKE_CASE__:   Tuple=      sentence[0], sentence[1:-1], sentence[-1]
     SCREAMING_SNAKE_CASE__:   Tuple=      sentence[: self.max_seq_length]

     SCREAMING_SNAKE_CASE__:   Optional[Any]=      torch.zeros(self.n_classes   )
     SCREAMING_SNAKE_CASE__:   Dict=      1

     SCREAMING_SNAKE_CASE__:   Optional[int]=      Image.open(os.path.join(self.data_dir , self.data[index]['''img''']   )   ).convert('''RGB'''   )
     SCREAMING_SNAKE_CASE__:   int=      self.transforms(lowerCAmelCase   )

     return {
         "image_start_token": start_token,
         "image_end_token": end_token,
         "sentence": sentence,
         "image": image,
         "label": label,
     }



    def 						UpperCamelCase_				(  self   )       ->					Optional[int]:
     SCREAMING_SNAKE_CASE__:   List[Any]=      Counter()
     for row in self.data:
      label_freqs.update(row['''label''']   )
     return label_freqs







def 			A__      (	snake_case_ :		int      ):
 SCREAMING_SNAKE_CASE__:   Any=      [len(row['''sentence''']      ) for row in batch]
 SCREAMING_SNAKE_CASE__,	SCREAMING_SNAKE_CASE__:   Optional[int]=      len(snake_case_      ), max(snake_case_      )

 SCREAMING_SNAKE_CASE__:   Union[str, Any]=      torch.zeros(snake_case_       ,				snake_case_       ,				dtype=torch.long      )
 SCREAMING_SNAKE_CASE__:   List[Any]=      torch.zeros(snake_case_       ,				snake_case_       ,				dtype=torch.long      )

 for i_batch, (input_row, length) in enumerate(zip(snake_case_       ,				snake_case_      )      ):
  SCREAMING_SNAKE_CASE__:   Optional[int]=      input_row['''sentence''']
  SCREAMING_SNAKE_CASE__:   Dict=      1

 SCREAMING_SNAKE_CASE__:   Optional[Any]=      torch.stack([row['''image'''] for row in batch]      )
 SCREAMING_SNAKE_CASE__:   List[str]=      torch.stack([row['''label'''] for row in batch]      )
 SCREAMING_SNAKE_CASE__:   Any=      torch.stack([row['''image_start_token'''] for row in batch]      )
 SCREAMING_SNAKE_CASE__:   Optional[int]=      torch.stack([row['''image_end_token'''] for row in batch]      )

 return text_tensor, mask_tensor, img_tensor, img_start_token, img_end_token, tgt_tensor







def 			A__      (	):
 return [
     "Crime",
     "Drama",
     "Thriller",
     "Action",
     "Comedy",
     "Romance",
     "Documentary",
     "Short",
     "Mystery",
     "History",
     "Family",
     "Adventure",
     "Fantasy",
     "Sci-Fi",
     "Western",
     "Horror",
     "Sport",
     "War",
     "Music",
     "Musical",
     "Animation",
     "Biography",
     "Film-Noir",
 ]







def 			A__      (	):
 return transforms.Compose(
     [
         transforms.Resize(256      ),
         transforms.CenterCrop(224      ),
         transforms.ToTensor(),
         transforms.Normalize(
             mean=[0.46_77_70_44, 0.44_53_14_29, 0.40_66_10_17]       ,				std=[0.12_22_19_94, 0.12_14_58_35, 0.14_38_04_69]       ,				),
     ]      )

