





"""simple docstring"""





import inspect
from typing import Callable, List, Optional, Union

import torch
from transformers import CLIPImageProcessor, CLIPTextModel, CLIPTokenizer

from diffusers import DiffusionPipeline
from diffusers.models import AutoencoderKL, UNetaDConditionModel
from diffusers.pipelines.stable_diffusion import StableDiffusionPipelineOutput
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker
from diffusers.schedulers import DDIMScheduler, LMSDiscreteScheduler, PNDMScheduler
from diffusers.utils import logging


lowercase__				:Union[str, Any]				   =       logging.get_logger(__name__)  # pylint: disable=invalid-name




class 							snake_case							(			__UpperCAmelCase ):

      '''simple docstring'''



      def __init__(							self       :      Optional[Any]						,				__lowercase       :      AutoencoderKL						,				__lowercase       :      CLIPTextModel						,				__lowercase       :      CLIPTokenizer						,				__lowercase       :      UNetaDConditionModel						,				__lowercase       :      Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler]						,				__lowercase       :      StableDiffusionSafetyChecker						,				__lowercase       :      CLIPImageProcessor						,				):


        '''simple docstring'''




        super().__init__()
        self.register_modules(
            vae=__lowercase						,				text_encoder=__lowercase						,				tokenizer=__lowercase						,				unet=__lowercase						,				scheduler=__lowercase						,				safety_checker=__lowercase						,				feature_extractor=__lowercase						,				)

      def       A_					(							self       :      Union[str, Any]						,				__lowercase       :      Optional[Union[str, int]] = "auto" ):


        '''simple docstring'''




        if slice_size == "auto":
          # half the attention head size is usually a good trade-off between
          # speed and memory
          __UpperCAmelCase    :						Any       =						self.unet.config.attention_head_dim // 2
        self.unet.set_attention_slice(__lowercase )

      def       A_					(							self       :      Tuple ):


        '''simple docstring'''




        self.enable_attention_slicing(__lowercase )







      @torch.no_grad()
      def __call__(							self       :      str						,				__lowercase       :      Union[str, List[str]]						,				__lowercase       :      int = 512						,				__lowercase       :      int = 512						,				__lowercase       :      int = 50						,				__lowercase       :      float = 7.5						,				__lowercase       :      Optional[Union[str, List[str]]] = None						,				__lowercase       :      Optional[int] = 1						,				__lowercase       :      float = 0.0						,				__lowercase       :      Optional[torch.Generator] = None						,				__lowercase       :      Optional[torch.FloatTensor] = None						,				__lowercase       :      Optional[str] = "pil"						,				__lowercase       :      bool = True						,				__lowercase       :      Optional[Callable[[int, int, torch.FloatTensor], None]] = None						,				__lowercase       :      int = 1						,				__lowercase       :      Optional[torch.FloatTensor] = None						,				**__lowercase       :      Any						,				):


        '''simple docstring'''




        if isinstance(__lowercase						,				__lowercase ):
          __UpperCAmelCase    :						Any       =						1
        elif isinstance(__lowercase						,				__lowercase ):
          __UpperCAmelCase    :						List[str]       =						len(__lowercase )
        else:
          raise ValueError(f'''`prompt` has to be of type `str` or `list` but is {type(__lowercase )}''' )

        if height % 8 != 0 or width % 8 != 0:
          raise ValueError(f'''`height` and `width` have to be divisible by 8 but are {height} and {width}.''' )

        if (callback_steps is None) or (
            callback_steps is not None and (not isinstance(__lowercase						,				__lowercase ) or callback_steps <= 0)
        ):
          raise ValueError(
              f'''`callback_steps` has to be a positive integer but is {callback_steps} of type'''
              f''' {type(__lowercase )}.''' )

        # get prompt text embeddings
        __UpperCAmelCase    :						List[Any]       =						self.tokenizer(
            __lowercase						,				padding='''max_length'''						,				max_length=self.tokenizer.model_max_length						,				return_tensors='''pt'''						,				)
        __UpperCAmelCase    :						Any       =						text_inputs.input_ids

        if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
          __UpperCAmelCase    :						Optional[int]       =						self.tokenizer.batch_decode(text_input_ids[:, self.tokenizer.model_max_length :] )
          logger.warning(
              '''The following part of your input was truncated because CLIP can only handle sequences up to'''
              f''' {self.tokenizer.model_max_length} tokens: {removed_text}''' )
          __UpperCAmelCase    :						Optional[Any]       =						text_input_ids[:, : self.tokenizer.model_max_length]

        if text_embeddings is None:
          __UpperCAmelCase    :						List[str]       =						self.text_encoder(text_input_ids.to(self.device ) )[0]

        # duplicate text embeddings for each generation per prompt, using mps friendly method
        __UpperCAmelCase    :						str       =						text_embeddings.shape
        __UpperCAmelCase    :						List[str]       =						text_embeddings.repeat(1						,				__lowercase						,				1 )
        __UpperCAmelCase    :						Optional[int]       =						text_embeddings.view(bs_embed * num_images_per_prompt						,				__lowercase						,				-1 )

        # here `guidance_scale` is defined analog to the guidance weight `w` of equation (2)
        # of the Imagen paper: https://arxiv.org/pdf/2205.11487.pdf . `guidance_scale = 1`
        # corresponds to doing no classifier free guidance.
        __UpperCAmelCase    :						List[Any]       =						guidance_scale > 1.0
        # get unconditional embeddings for classifier free guidance
        if do_classifier_free_guidance:
          __UpperCAmelCase    :						List[str]
          if negative_prompt is None:
            __UpperCAmelCase    :						Optional[int]       =						['''''']
          elif type(__lowercase ) is not type(__lowercase ):
            raise TypeError(
                f'''`negative_prompt` should be the same type to `prompt`, but got {type(__lowercase )} !='''
                f''' {type(__lowercase )}.''' )
          elif isinstance(__lowercase						,				__lowercase ):
            __UpperCAmelCase    :						Any       =						[negative_prompt]
          elif batch_size != len(__lowercase ):
            raise ValueError(
                f'''`negative_prompt`: {negative_prompt} has batch size {len(__lowercase )}, but `prompt`:'''
                f''' {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches'''
                ''' the batch size of `prompt`.''' )
          else:
            __UpperCAmelCase    :						Tuple       =						negative_prompt

          __UpperCAmelCase    :						Optional[Any]       =						text_input_ids.shape[-1]
          __UpperCAmelCase    :						Tuple       =						self.tokenizer(
              __lowercase						,				padding='''max_length'''						,				max_length=__lowercase						,				truncation=__lowercase						,				return_tensors='''pt'''						,				)
          __UpperCAmelCase    :						int       =						self.text_encoder(uncond_input.input_ids.to(self.device ) )[0]

          # duplicate unconditional embeddings for each generation per prompt, using mps friendly method
          __UpperCAmelCase    :						Union[str, Any]       =						uncond_embeddings.shape[1]
          __UpperCAmelCase    :						Optional[int]       =						uncond_embeddings.repeat(__lowercase						,				__lowercase						,				1 )
          __UpperCAmelCase    :						Union[str, Any]       =						uncond_embeddings.view(batch_size * num_images_per_prompt						,				__lowercase						,				-1 )

          # For classifier free guidance, we need to do two forward passes.
          # Here we concatenate the unconditional and text embeddings into a single batch
          # to avoid doing two forward passes
          __UpperCAmelCase    :						List[Any]       =						torch.cat([uncond_embeddings, text_embeddings] )

        # get the initial random noise unless the user supplied it

        # Unlike in other pipelines, latents need to be generated in the target device
        # for 1-to-1 results reproducibility with the CompVis implementation.
        # However this currently doesn't work in `mps`.
        __UpperCAmelCase    :						Dict       =						(batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
        __UpperCAmelCase    :						str       =						(batch_size * num_images_per_prompt, self.unet.config.in_channels, 64, 64)
        __UpperCAmelCase    :						int       =						text_embeddings.dtype
        if latents is None:
          if self.device.type == "mps":
            # randn does not exist on mps
            __UpperCAmelCase    :						List[Any]       =						torch.randn(
                __lowercase						,				generator=__lowercase						,				device='''cpu'''						,				dtype=__lowercase ).to(self.device )
            __UpperCAmelCase    :						List[Any]       =						torch.randn(__lowercase						,				generator=__lowercase						,				device='''cpu'''						,				dtype=__lowercase ).to(
                self.device )
          else:
            __UpperCAmelCase    :						List[str]       =						torch.randn(
                __lowercase						,				generator=__lowercase						,				device=self.device						,				dtype=__lowercase )
            __UpperCAmelCase    :						Dict       =						torch.randn(__lowercase						,				generator=__lowercase						,				device=self.device						,				dtype=__lowercase )
        else:
          if latents_reference.shape != latents_shape:
            raise ValueError(f'''Unexpected latents shape, got {latents.shape}, expected {latents_shape}''' )
          __UpperCAmelCase    :						List[Any]       =						latents_reference.to(self.device )
          __UpperCAmelCase    :						Optional[Any]       =						latents.to(self.device )

        # This is the key part of the pipeline where we
        # try to ensure that the generated images w/ the same seed
        # but different sizes actually result in similar images
        __UpperCAmelCase    :						Optional[Any]       =						(latents_shape[3] - latents_shape_reference[3]) // 2
        __UpperCAmelCase    :						Any       =						(latents_shape[2] - latents_shape_reference[2]) // 2
        __UpperCAmelCase    :						List[str]       =						latents_shape_reference[3] if dx >= 0 else latents_shape_reference[3] + 2 * dx
        __UpperCAmelCase    :						int       =						latents_shape_reference[2] if dy >= 0 else latents_shape_reference[2] + 2 * dy
        __UpperCAmelCase    :						int       =						0 if dx < 0 else dx
        __UpperCAmelCase    :						Optional[int]       =						0 if dy < 0 else dy
        __UpperCAmelCase    :						List[Any]       =						max(-dx						,				0 )
        __UpperCAmelCase    :						int       =						max(-dy						,				0 )
        # import pdb
        # pdb.set_trace()
        __UpperCAmelCase    :						Optional[int]       =						latents_reference[:, :, dy : dy + h, dx : dx + w]

        # set timesteps
        self.scheduler.set_timesteps(__lowercase )

        # Some schedulers like PNDM have timesteps as arrays
        # It's more optimized to move all timesteps to correct device beforehand
        __UpperCAmelCase    :						Optional[Any]       =						self.scheduler.timesteps.to(self.device )

        # scale the initial noise by the standard deviation required by the scheduler
        __UpperCAmelCase    :						int       =						latents * self.scheduler.init_noise_sigma

        # prepare extra kwargs for the scheduler step, since not all schedulers have the same signature
        # eta (η) is only used with the DDIMScheduler, it will be ignored for other schedulers.
        # eta corresponds to η in DDIM paper: https://arxiv.org/abs/2010.02502
        # and should be between [0, 1]
        __UpperCAmelCase    :						str       =						'''eta''' in set(inspect.signature(self.scheduler.step ).parameters.keys() )
        __UpperCAmelCase    :						int       =						{}
        if accepts_eta:
          __UpperCAmelCase    :						List[Any]       =						eta

        for i, t in enumerate(self.progress_bar(__lowercase ) ):
          # expand the latents if we are doing classifier free guidance
          __UpperCAmelCase    :						int       =						torch.cat([latents] * 2 ) if do_classifier_free_guidance else latents
          __UpperCAmelCase    :						Optional[int]       =						self.scheduler.scale_model_input(__lowercase						,				__lowercase )

          # predict the noise residual
          __UpperCAmelCase    :						Union[str, Any]       =						self.unet(__lowercase						,				__lowercase						,				encoder_hidden_states=__lowercase ).sample

          # perform guidance
          if do_classifier_free_guidance:
            __UpperCAmelCase    :						Dict       =						noise_pred.chunk(2 )
            __UpperCAmelCase    :						Optional[int]       =						noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

          # compute the previous noisy sample x_t -> x_t-1
          __UpperCAmelCase    :						int       =						self.scheduler.step(__lowercase						,				__lowercase						,				__lowercase						,				**__lowercase ).prev_sample

          # call the callback, if provided
          if callback is not None and i % callback_steps == 0:
            callback(__lowercase						,				__lowercase						,				__lowercase )

        __UpperCAmelCase    :						Optional[int]       =						1 / 0.1_8_2_1_5 * latents
        __UpperCAmelCase    :						str       =						self.vae.decode(__lowercase ).sample

        __UpperCAmelCase    :						str       =						(image / 2 + 0.5).clamp(0						,				1 )

        # we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
        __UpperCAmelCase    :						int       =						image.cpu().permute(0						,				2						,				3						,				1 ).float().numpy()

        if self.safety_checker is not None:
          __UpperCAmelCase    :						str       =						self.feature_extractor(self.numpy_to_pil(__lowercase )						,				return_tensors='''pt''' ).to(
              self.device )
          __UpperCAmelCase    :						Dict       =						self.safety_checker(
              images=__lowercase						,				clip_input=safety_checker_input.pixel_values.to(text_embeddings.dtype ) )
        else:
          __UpperCAmelCase    :						Union[str, Any]       =						None

        if output_type == "pil":
          __UpperCAmelCase    :						Any       =						self.numpy_to_pil(__lowercase )

        if not return_dict:
          return (image, has_nsfw_concept)

        return StableDiffusionPipelineOutput(images=__lowercase						,				nsfw_content_detected=__lowercase )






"""simple docstring"""





import argparse

import torch

from transformers import LxmertConfig, LxmertForPreTraining, load_tf_weights_in_lxmert
from transformers.utils import logging


logging.set_verbosity_info()


def       lowerCamelCase_       (   UpperCAmelCase_  ,			UpperCAmelCase_  ,			UpperCAmelCase_  )						->Optional[int]:





  """simple docstring"""


  __UpperCAmelCase    :						Dict       =						LxmertConfig.from_json_file(UpperCAmelCase_  )
  print(f'''Building PyTorch model from configuration: {config}'''  )
  __UpperCAmelCase    :						Tuple       =						LxmertForPreTraining(UpperCAmelCase_  )

  # Load weights from tf checkpoint
  load_tf_weights_in_lxmert(UpperCAmelCase_  ,			UpperCAmelCase_  ,			UpperCAmelCase_  )

  # Save pytorch-model
  print(f'''Save PyTorch model to {pytorch_dump_path}'''  )
  torch.save(model.state_dict()  ,			UpperCAmelCase_  )


if __name__ == "__main__":
  lowercase__				:Optional[int]				   =       argparse.ArgumentParser()
  # Required parameters
  parser.add_argument(
      '--tf_checkpoint_path', default=None, type=str, required=True, help='Path to the TensorFlow checkpoint path.'
  )
  parser.add_argument(
      '--config_file',
      default=None,
      type=str,
      required=True,
      help='The config json file corresponding to the pre-trained model. \nThis specifies the model architecture.',
  )
  parser.add_argument(
      '--pytorch_dump_path', default=None, type=str, required=True, help='Path to the output PyTorch model.'
  )
  lowercase__				:Union[str, Any]				   =       parser.parse_args()
  convert_tf_checkpoint_to_pytorch(args.tf_checkpoint_path, args.config_file, args.pytorch_dump_path)