




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__	( UpperCamelCase__       , UpperCamelCase__       , UpperCamelCase__       , unittest.TestCase				):
     a		:      Optional[int]		 =     StableUnCLIPImgaImgPipeline
     a		:      List[Any]		 =     TEXT_GUIDED_IMAGE_VARIATION_PARAMS
     a		:      Tuple		 =     TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
     a		:      Union[str, Any]		 =     frozenset(
         []				)  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
     a		:      Tuple		 =     frozenset([]				)







     def       lowerCAmelCase_    (     self			)  ->   Tuple:



          '''simple docstring'''



          a     =				32
          a     =				embedder_hidden_size

          # image encoding components

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

          torch.manual_seed(0			)
          a     =				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			)
          a     =				StableUnCLIPImageNormalizer(embedding_dim=lowerCamelCase_			)
          a     =				DDPMScheduler(beta_schedule="squaredcos_cap_v2"			)

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

          torch.manual_seed(0			)
          a     =				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			)
          a     =				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			)
          a     =				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=lowerCamelCase_  ,		steps_offset=1  ,		)

          torch.manual_seed(0			)
          a     =				AutoencoderKL()

          a     =				{
              # 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       lowerCAmelCase_    (     self  ,		A  ,		A=0  ,		A=True			)  ->   Optional[Any]:



          '''simple docstring'''



          if str(lowerCamelCase_			).startswith("mps"			):
               a     =				torch.manual_seed(lowerCamelCase_			)
          else:
               a     =				torch.Generator(device=lowerCamelCase_			).manual_seed(lowerCamelCase_			)

          a     =				floats_tensor((1, 3, 32, 32)  ,		rng=random.Random(lowerCamelCase_			)			).to(lowerCamelCase_			)

          if pil_image:
               a     =				input_image * 0.5 + 0.5
               a     =				input_image.clamp(0  ,		1			)
               a     =				input_image.cpu().permute(0  ,		2  ,		3  ,		1			).float().numpy()
               a     =				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       lowerCAmelCase_    (     self			)  ->   List[Any]:



          '''simple docstring'''



          a     =				"cpu"  # ensure determinism for the device-dependent torch.Generator
          a     =				self.get_dummy_components()
          a     =				StableUnCLIPImgaImgPipeline(**lowerCamelCase_			)
          a     =				sd_pipe.to(lowerCamelCase_			)
          sd_pipe.set_progress_bar_config(disable=lowerCamelCase_			)

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

          assert image.shape == (1, 32, 32, 3)
          a     =				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       lowerCAmelCase_    (     self			)  ->   List[Any]:



          '''simple docstring'''



          a     =				torch_device in ["cpu", "mps"]

          self._test_attention_slicing_forward_pass(test_max_difference=lowerCamelCase_			)







     def       lowerCAmelCase_    (     self			)  ->   Tuple:



          '''simple docstring'''



          a     =				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       lowerCAmelCase_    (     self			)  ->   Dict:



          '''simple docstring'''



          self._test_xformers_attention_forwardGenerator_pass(test_max_difference=lowerCamelCase_			)







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







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



          '''simple docstring'''



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







     def       lowerCAmelCase_    (     self			)  ->   int:



          '''simple docstring'''



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

          a     =				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"			)

          a     =				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()

          a     =				torch.Generator(device="cpu"			).manual_seed(0			)
          a     =				pipe(lowerCamelCase_  ,		"anime turle"  ,		generator=lowerCamelCase_  ,		output_type="np"			)

          a     =				output.images[0]

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

          assert_mean_pixel_difference(lowerCamelCase_  ,		lowerCamelCase_			)







     def       lowerCAmelCase_    (     self			)  ->   Dict:



          '''simple docstring'''



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

          a     =				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"			)

          a     =				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()

          a     =				torch.Generator(device="cpu"			).manual_seed(0			)
          a     =				pipe(lowerCamelCase_  ,		"anime turle"  ,		generator=lowerCamelCase_  ,		output_type="np"			)

          a     =				output.images[0]

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

          assert_mean_pixel_difference(lowerCamelCase_  ,		lowerCamelCase_			)







     def       lowerCAmelCase_    (     self			)  ->   Optional[Any]:



          '''simple docstring'''



          a     =				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()

          a     =				StableUnCLIPImgaImgPipeline.from_pretrained(
              "fusing/stable-unclip-2-1-h-img2img"  ,		torch_dtype=torch.floataa			)
          a     =				pipe.to(lowerCamelCase_			)
          pipe.set_progress_bar_config(disable=lowerCamelCase_			)
          pipe.enable_attention_slicing()
          pipe.enable_sequential_cpu_offload()

          a     =				pipe(
              lowerCamelCase_  ,		"anime turtle"  ,		num_inference_steps=2  ,		output_type="np"  ,		)

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









import argparse

import pytorch_lightning as pl
import torch
from torch import nn

from transformers import LongformerForQuestionAnswering, LongformerModel




class       a__	( pl.LightningModule				):







     def __init__(     self  ,		A			)  ->   Optional[Any]:



          '''simple docstring'''



          super().__init__()
          a     =				model
          a     =				2
          a     =				nn.Linear(self.model.config.hidden_size  ,		self.num_labels			)







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



          '''simple docstring'''



          pass







def  SCREAMING_SNAKE_CASE							(       __UpperCamelCase		, __UpperCamelCase		, __UpperCamelCase)      ->					List[Any]:
     # load longformer model from model identifier
     a     =				LongformerModel.from_pretrained(__UpperCamelCase)
     a     =				LightningModel(__UpperCamelCase)

     a     =				torch.load(__UpperCamelCase		, map_location=torch.device("cpu"))
     lightning_model.load_state_dict(ckpt["state_dict"])

     # init longformer question answering model
     a     =				LongformerForQuestionAnswering.from_pretrained(__UpperCamelCase)

     # transfer weights
     longformer_for_qa.longformer.load_state_dict(lightning_model.model.state_dict())
     longformer_for_qa.qa_outputs.load_state_dict(lightning_model.qa_outputs.state_dict())
     longformer_for_qa.eval()

     # save model
     longformer_for_qa.save_pretrained(__UpperCamelCase)

     print(f'''Conversion successful. Model saved under {pytorch_dump_folder_path}''')


if __name__ == "__main__":
   lowercase__				:      Optional[Any]	   = argparse.ArgumentParser()
   # Required parameters
   parser.add_argument(
       "--longformer_model",
       default=None,
       type=str,
       required=True,
       help="model identifier of longformer. Should be either `longformer-base-4096` or `longformer-large-4096`.",
   )
   parser.add_argument(
       "--longformer_question_answering_ckpt_path",
       default=None,
       type=str,
       required=True,
       help="Path the official PyTorch Lightning Checkpoint.",
   )
   parser.add_argument(
       "--pytorch_dump_folder_path", default=None, type=str, required=True, help="Path to the output PyTorch model."
   )
   lowercase__				:      Union[str, Any]	   = parser.parse_args()
   convert_longformer_qa_checkpoint_to_pytorch(
       args.longformer_model, args.longformer_question_answering_ckpt_path, args.pytorch_dump_folder_path
   )

