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






 '''simple docstring'''






 UpperCAmelCase_     			= StableUnCLIPImgaImgPipeline
 UpperCAmelCase_     			= TEXT_GUIDED_IMAGE_VARIATION_PARAMS
 UpperCAmelCase_     			= TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
 UpperCAmelCase_     			= frozenset(
     []						)  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
 UpperCAmelCase_     			= frozenset([]						)




 def 		__snake_case						(      self :  List[str]):







        '''simple docstring'''



        lowerCAmelCase__					     = 32
        lowerCAmelCase__					     = embedder_hidden_size

        # image encoding components

        lowerCAmelCase__					     = CLIPImageProcessor(crop_size=32      ,   size=32)

        torch.manual_seed(0)
        lowerCAmelCase__					     = CLIPVisionModelWithProjection(
            CLIPVisionConfig(
                hidden_size=lowercase__      ,   projection_dim=lowercase__      ,   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__					     = StableUnCLIPImageNormalizer(embedding_dim=lowercase__)
        lowerCAmelCase__					     = DDPMScheduler(beta_schedule='squaredcos_cap_v2')

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

        torch.manual_seed(0)
        lowerCAmelCase__					     = CLIPTextModel(
            CLIPTextConfig(
                bos_token_id=0      ,   eos_token_id=2      ,   hidden_size=lowercase__      ,   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=1_000      ,   ))

        torch.manual_seed(0)
        lowerCAmelCase__					     = 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=lowercase__      ,   layers_per_block=1      ,   upcast_attention=lowercase__      ,   use_linear_projection=lowercase__      ,   )

        torch.manual_seed(0)
        lowerCAmelCase__					     = DDIMScheduler(
            beta_schedule='scaled_linear'      ,   beta_start=0.00_085      ,   beta_end=0.012      ,   prediction_type='v_prediction'      ,   set_alpha_to_one=lowercase__      ,   steps_offset=1      ,   )

        torch.manual_seed(0)
        lowerCAmelCase__					     = AutoencoderKL()

        lowerCAmelCase__					     = {
            # 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 		__snake_case						(      self :  Tuple      ,   lowercase__ :  Tuple      ,   lowercase__ :  Union[str, Any]=0      ,   lowercase__ :  Tuple=True):







        '''simple docstring'''



        if str(lowercase__).startswith('mps'):
               lowerCAmelCase__					     = torch.manual_seed(lowercase__)
        else:
               lowerCAmelCase__					     = torch.Generator(device=lowercase__).manual_seed(lowercase__)

        lowerCAmelCase__					     = floats_tensor((1, 3, 32, 32)      ,   rng=random.Random(lowercase__)).to(lowercase__)

        if pil_image:
               lowerCAmelCase__					     = input_image * 0.5 + 0.5
               lowerCAmelCase__					     = input_image.clamp(0      ,   1)
               lowerCAmelCase__					     = input_image.cpu().permute(0      ,   2      ,   3      ,   1).float().numpy()
               lowerCAmelCase__					     = DiffusionPipeline.numpy_to_pil(lowercase__)[0]

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




 @skip_mps
 def 		__snake_case						(      self :  Optional[int]):







        '''simple docstring'''



        lowerCAmelCase__					     = 'cpu'  # ensure determinism for the device-dependent torch.Generator
        lowerCAmelCase__					     = self.get_dummy_components()
        lowerCAmelCase__					     = StableUnCLIPImgaImgPipeline(**lowercase__)
        lowerCAmelCase__					     = sd_pipe.to(lowercase__)
        sd_pipe.set_progress_bar_config(disable=lowercase__)

        lowerCAmelCase__					     = self.get_dummy_inputs(lowercase__)
        inputs.update({'image_embeds': None})
        lowerCAmelCase__					     = sd_pipe(**lowercase__).images
        lowerCAmelCase__					     = image[0, -3:, -3:, -1]

        assert image.shape == (1, 32, 32, 3)
        lowerCAmelCase__					     = np.array([0.3_872, 0.7_224, 0.5_601, 0.4_741, 0.6_872, 0.5_814, 0.4_636, 0.3_867, 0.5_078])

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




 def 		__snake_case						(      self :  Tuple):







        '''simple docstring'''



        lowerCAmelCase__					     = torch_device in ['cpu', 'mps']

        self._test_attention_slicing_forward_pass(test_max_difference=lowercase__)




 def 		__snake_case						(      self :  Tuple):







        '''simple docstring'''



        lowerCAmelCase__					     = torch_device in ['cpu', 'mps']

        self._test_inference_batch_single_identical(test_max_difference=lowercase__)




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







        '''simple docstring'''



        self._test_xformers_attention_forwardGenerator_pass(test_max_difference=lowercase__)






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






 '''simple docstring'''






 def 		__snake_case						(      self :  Union[str, Any]):







        '''simple docstring'''



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




 def 		__snake_case						(      self :  Optional[Any]):







        '''simple docstring'''



        lowerCAmelCase__					     = load_image(
            'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png')

        lowerCAmelCase__					     = 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__					     = StableUnCLIPImgaImgPipeline.from_pretrained(
            'fusing/stable-unclip-2-1-l-img2img'      ,   torch_dtype=torch.floataa)
        pipe.to(lowercase__)
        pipe.set_progress_bar_config(disable=lowercase__)
        # 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__					     = torch.Generator(device='cpu').manual_seed(0)
        lowerCAmelCase__					     = pipe(lowercase__      ,   'anime turle'      ,   generator=lowercase__      ,   output_type='np')

        lowerCAmelCase__					     = output.images[0]

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

        assert_mean_pixel_difference(lowercase__      ,   lowercase__)




 def 		__snake_case						(      self :  List[Any]):







        '''simple docstring'''



        lowerCAmelCase__					     = load_image(
            'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png')

        lowerCAmelCase__					     = 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__					     = StableUnCLIPImgaImgPipeline.from_pretrained(
            'fusing/stable-unclip-2-1-h-img2img'      ,   torch_dtype=torch.floataa)
        pipe.to(lowercase__)
        pipe.set_progress_bar_config(disable=lowercase__)
        # 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__					     = torch.Generator(device='cpu').manual_seed(0)
        lowerCAmelCase__					     = pipe(lowercase__      ,   'anime turle'      ,   generator=lowercase__      ,   output_type='np')

        lowerCAmelCase__					     = output.images[0]

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

        assert_mean_pixel_difference(lowercase__      ,   lowercase__)




 def 		__snake_case						(      self :  Optional[Any]):







        '''simple docstring'''



        lowerCAmelCase__					     = 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__					     = StableUnCLIPImgaImgPipeline.from_pretrained(
            'fusing/stable-unclip-2-1-h-img2img'      ,   torch_dtype=torch.floataa)
        lowerCAmelCase__					     = pipe.to(lowercase__)
        pipe.set_progress_bar_config(disable=lowercase__)
        pipe.enable_attention_slicing()
        pipe.enable_sequential_cpu_offload()

        lowerCAmelCase__					     = pipe(
            lowercase__      ,   'anime turtle'      ,   num_inference_steps=2      ,   output_type='np'      ,   )

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

import argparse
from collections import defaultdict

import yaml


lowerCAmelCase__     						=			'docs/source/en/_toctree.yml'






def 				__lowerCamelCase		(						lowerCAmelCase__     ):
       lowerCAmelCase__					     = defaultdict(lowerCAmelCase__     )
       for doc in model_doc:
              counts[doc["local"]] += 1
       lowerCAmelCase__					     = [key for key, value in counts.items() if value > 1]

       lowerCAmelCase__					     = []
       for duplicate_key in duplicates:
              lowerCAmelCase__					     = list({doc['title'] for doc in model_doc if doc['local'] == duplicate_key}     )
              if len(lowerCAmelCase__     ) > 1:
                     raise ValueError(
                         F"""{duplicate_key} is present several times in the documentation table of content at """
                         '`docs/source/en/_toctree.yml` with different *Title* values. Choose one of those and remove the '
                         'others.'     )
              # Only add this once
              new_doc.append({'local': duplicate_key, 'title': titles[0]}     )

       # Add none duplicate-keys
       new_doc.extend([doc for doc in model_doc if counts[doc['local']] == 1]     )

       # Sort
       return sorted(lowerCAmelCase__     ,  key=lambda lowerCAmelCase__     : s["title"].lower()     )






def 				__lowerCamelCase		(						lowerCAmelCase__=False     ):
       with open(lowerCAmelCase__     ,  encoding='utf-8'     ) as f:
              lowerCAmelCase__					     = yaml.safe_load(f.read()     )

       # Get to the API doc
       lowerCAmelCase__					     = 0
       while content[api_idx]["title"] != "API":
              api_idx += 1
       lowerCAmelCase__					     = content[api_idx]['sections']

       # Then to the model doc
       lowerCAmelCase__					     = 0
       while api_doc[model_idx]["title"] != "Models":
              model_idx += 1

       lowerCAmelCase__					     = api_doc[model_idx]['sections']

       lowerCAmelCase__					     = [(idx, section) for idx, section in enumerate(lowerCAmelCase__     ) if 'sections' in section]
       lowerCAmelCase__					     = False
       for idx, modality_doc in modalities_docs:
              lowerCAmelCase__					     = modality_doc['sections']
              lowerCAmelCase__					     = clean_model_doc_toc(lowerCAmelCase__     )

              if old_modality_doc != new_modality_doc:
                     lowerCAmelCase__					     = True
                     if overwrite:
                            lowerCAmelCase__					     = new_modality_doc

       if diff:
              if overwrite:
                     lowerCAmelCase__					     = model_doc
                     lowerCAmelCase__					     = api_doc
                     with open(lowerCAmelCase__     ,  'w'     ,  encoding='utf-8'     ) as f:
                            f.write(yaml.dump(lowerCAmelCase__     ,  allow_unicode=lowerCAmelCase__     )     )
              else:
                     raise ValueError(
                         'The model doc part of the table of content is not properly sorted, run `make style` to fix this.'     )


if __name__ == "__main__":
    lowerCAmelCase__     						=			argparse.ArgumentParser()
    parser.add_argument('--fix_and_overwrite', action='store_true', help='Whether to fix inconsistencies.')
    lowerCAmelCase__     						=			parser.parse_args()

    check_model_doc(args.fix_and_overwrite)
