





"""simple docstring"""




import types
from typing import List, Optional, Tuple, Union

import torch
from transformers import CLIPTextModelWithProjection, CLIPTokenizer
from transformers.models.clip.modeling_clip import CLIPTextModelOutput

from diffusers.models import PriorTransformer
from diffusers.pipelines import DiffusionPipeline, StableDiffusionImageVariationPipeline
from diffusers.schedulers import UnCLIPScheduler
from diffusers.utils import logging, randn_tensor


lowerCamelCase_					:						List[str]      = logging.get_logger(__name__)  # pylint: disable=invalid-name





def 						UpperCAmelCase__		(					self			, _UpperCAmelCase			, _UpperCAmelCase			, _UpperCAmelCase			, _UpperCAmelCase ):






  """simple docstring"""





  A_  :	Dict    						=  image.to(device=_SCREAMING_SNAKE_CASE )
  A_  :	str    						=  image  # take image as image_embeddings
  A_  :	List[Any]    						=  image_embeddings.unsqueeze(1 )

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

  if do_classifier_free_guidance:
    A_  :	Dict    						=  torch.zeros_like(_SCREAMING_SNAKE_CASE )

    # 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
    A_  :	Optional[Any]    						=  torch.cat([uncond_embeddings, image_embeddings] )

  return image_embeddings





class 			_UpperCAmelCase  (       lowerCAmelCase__		):



      '''simple docstring'''

      def __init__(					self ,				snake_case_ ,				snake_case_ ,				snake_case_ ,				snake_case_ ,				snake_case_ = None ,				):

        """simple docstring"""

        super().__init__()

        A_  :	str    						=  {"image_encoder": None} if decoder_pipe_kwargs is None else decoder_pipe_kwargs

        A_  :	Dict    						=  decoder_pipe_kwargs.get('torch_dtype' ,				UpperCamelCase__  ) or prior.dtype

        A_  :	Dict    						=  StableDiffusionImageVariationPipeline.from_pretrained(
            'lambdalabs/sd-image-variations-diffusers' ,				**UpperCamelCase__  )

        # replace `_encode_image` method
        A_  :	List[str]    						=  types.MethodType(_encode_image ,				self.decoder_pipe  )

        self.register_modules(
            prior=UpperCamelCase__ ,				tokenizer=UpperCamelCase__ ,				text_encoder=UpperCamelCase__ ,				prior_scheduler=UpperCamelCase__ ,				)

      def  lowerCamelCase_      (					self ,				snake_case_ ,				snake_case_ ,				snake_case_ ,				snake_case_ ,				snake_case_ = None ,				snake_case_ = None ,				):

        """simple docstring"""

        if text_model_output is None:
          A_  :	List[str]    						=  len(UpperCamelCase__  ) if isinstance(UpperCamelCase__ ,				UpperCamelCase__  ) else 1
          # get prompt text embeddings
          A_  :	Any    						=  self.tokenizer(
              UpperCamelCase__ ,				padding='max_length' ,				max_length=self.tokenizer.model_max_length ,				return_tensors='pt' ,				)
          A_  :	int    						=  text_inputs.input_ids
          A_  :	Optional[Any]    						=  text_inputs.attention_mask.bool().to(UpperCamelCase__  )

          if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
            A_  :	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}"""  )
            A_  :	Optional[Any]    						=  text_input_ids[:, : self.tokenizer.model_max_length]

          A_  :	int    						=  self.text_encoder(text_input_ids.to(UpperCamelCase__  )  )

          A_  :	Optional[Any]    						=  text_encoder_output.text_embeds
          A_  :	Optional[int]    						=  text_encoder_output.last_hidden_state

        else:
          A_  :	Dict    						=  text_model_output[0].shape[0]
          A_  :	int    						=  text_model_output[0], text_model_output[1]
          A_  :	int    						=  text_attention_mask

        A_  :	Dict    						=  text_embeddings.repeat_interleave(UpperCamelCase__ ,				dim=0  )
        A_  :	int    						=  text_encoder_hidden_states.repeat_interleave(UpperCamelCase__ ,				dim=0  )
        A_  :	List[str]    						=  text_mask.repeat_interleave(UpperCamelCase__ ,				dim=0  )

        if do_classifier_free_guidance:
          A_  :	Dict    						=  [""] * batch_size

          A_  :	List[Any]    						=  self.tokenizer(
              UpperCamelCase__ ,				padding='max_length' ,				max_length=self.tokenizer.model_max_length ,				truncation=UpperCamelCase__ ,				return_tensors='pt' ,				)
          A_  :	str    						=  uncond_input.attention_mask.bool().to(UpperCamelCase__  )
          A_  :	int    						=  self.text_encoder(uncond_input.input_ids.to(UpperCamelCase__  )  )

          A_  :	Union[str, Any]    						=  uncond_embeddings_text_encoder_output.text_embeds
          A_  :	List[Any]    						=  uncond_embeddings_text_encoder_output.last_hidden_state

          # duplicate unconditional embeddings for each generation per prompt, using mps friendly method

          A_  :	Optional[Any]    						=  uncond_embeddings.shape[1]
          A_  :	str    						=  uncond_embeddings.repeat(1 ,				UpperCamelCase__  )
          A_  :	Optional[Any]    						=  uncond_embeddings.view(batch_size * num_images_per_prompt ,				UpperCamelCase__  )

          A_  :	Any    						=  uncond_text_encoder_hidden_states.shape[1]
          A_  :	int    						=  uncond_text_encoder_hidden_states.repeat(1 ,				UpperCamelCase__ ,				1  )
          A_  :	Tuple    						=  uncond_text_encoder_hidden_states.view(
              batch_size * num_images_per_prompt ,				UpperCamelCase__ ,				-1  )
          A_  :	List[str]    						=  uncond_text_mask.repeat_interleave(UpperCamelCase__ ,				dim=0  )

          # done duplicates

          # 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
          A_  :	int    						=  torch.cat([uncond_embeddings, text_embeddings]  )
          A_  :	str    						=  torch.cat([uncond_text_encoder_hidden_states, text_encoder_hidden_states]  )

          A_  :	Dict    						=  torch.cat([uncond_text_mask, text_mask]  )

        return text_embeddings, text_encoder_hidden_states, text_mask

      @property
      def  lowerCamelCase_      (					self  ):

        """simple docstring"""

        if self.device != torch.device('meta'  ) or not hasattr(self.prior ,				'_hf_hook'  ):
          return self.device
        for module in self.prior.modules():
          if (
              hasattr(UpperCamelCase__ ,				'_hf_hook'  )
              and hasattr(module._hf_hook ,				'execution_device'  )
              and module._hf_hook.execution_device is not None
          ):
            return torch.device(module._hf_hook.execution_device  )
        return self.device

      def  lowerCamelCase_      (					self ,				snake_case_ ,				snake_case_ ,				snake_case_ ,				snake_case_ ,				snake_case_ ,				snake_case_  ):

        """simple docstring"""

        if latents is None:
          A_  :	Any    						=  randn_tensor(UpperCamelCase__ ,				generator=UpperCamelCase__ ,				device=UpperCamelCase__ ,				dtype=UpperCamelCase__  )
        else:
          if latents.shape != shape:
            raise ValueError(F"""Unexpected latents shape, got {latents.shape}, expected {shape}"""  )
          A_  :	Any    						=  latents.to(UpperCamelCase__  )

        A_  :	List[str]    						=  latents * scheduler.init_noise_sigma
        return latents

      def  lowerCamelCase_      (					self ,				snake_case_ = None  ):

        """simple docstring"""

        self.decoder_pipe.to(UpperCamelCase__  )
        super().to(UpperCamelCase__  )

      @torch.no_grad()
      def __call__(					self ,				snake_case_ = None ,				snake_case_ = None ,				snake_case_ = None ,				snake_case_ = 1 ,				snake_case_ = 2_5 ,				snake_case_ = None ,				snake_case_ = None ,				snake_case_ = None ,				snake_case_ = None ,				snake_case_ = 4.0 ,				snake_case_ = 8.0 ,				snake_case_ = 5_0 ,				snake_case_ = 1 ,				snake_case_ = 0.0 ,				snake_case_ = "pil" ,				snake_case_ = True ,				):

        """simple docstring"""

        if prompt is not None:
          if isinstance(UpperCamelCase__ ,				UpperCamelCase__  ):
            A_  :	Union[str, Any]    						=  1
          elif isinstance(UpperCamelCase__ ,				UpperCamelCase__  ):
            A_  :	int    						=  len(UpperCamelCase__  )
          else:
            raise ValueError(F"""`prompt` has to be of type `str` or `list` but is {type(UpperCamelCase__  )}"""  )
        else:
          A_  :	Any    						=  text_model_output[0].shape[0]

        A_  :	str    						=  self._execution_device

        A_  :	Optional[Any]    						=  batch_size * num_images_per_prompt

        A_  :	int    						=  prior_guidance_scale > 1.0 or decoder_guidance_scale > 1.0

        A_  :	Tuple    						=  self._encode_prompt(
            UpperCamelCase__ ,				UpperCamelCase__ ,				UpperCamelCase__ ,				UpperCamelCase__ ,				UpperCamelCase__ ,				UpperCamelCase__  )

        # prior

        self.prior_scheduler.set_timesteps(UpperCamelCase__ ,				device=UpperCamelCase__  )
        A_  :	Optional[int]    						=  self.prior_scheduler.timesteps

        A_  :	List[Any]    						=  self.prior.config.embedding_dim

        A_  :	Dict    						=  self.prepare_latents(
            (batch_size, embedding_dim) ,				text_embeddings.dtype ,				UpperCamelCase__ ,				UpperCamelCase__ ,				UpperCamelCase__ ,				self.prior_scheduler ,				)

        for i, t in enumerate(self.progress_bar(UpperCamelCase__  )  ):
          # expand the latents if we are doing classifier free guidance
          A_  :	Tuple    						=  torch.cat([prior_latents] * 2  ) if do_classifier_free_guidance else prior_latents

          A_  :	List[Any]    						=  self.prior(
              UpperCamelCase__ ,				timestep=UpperCamelCase__ ,				proj_embedding=UpperCamelCase__ ,				encoder_hidden_states=UpperCamelCase__ ,				attention_mask=UpperCamelCase__ ,				).predicted_image_embedding

          if do_classifier_free_guidance:
            A_  :	List[Any]    						=  predicted_image_embedding.chunk(2  )
            A_  :	Optional[Any]    						=  predicted_image_embedding_uncond + prior_guidance_scale * (
                predicted_image_embedding_text - predicted_image_embedding_uncond
            )

          if i + 1 == prior_timesteps_tensor.shape[0]:
            A_  :	Union[str, Any]    						=  None
          else:
            A_  :	Dict    						=  prior_timesteps_tensor[i + 1]

          A_  :	Optional[int]    						=  self.prior_scheduler.step(
              UpperCamelCase__ ,				timestep=UpperCamelCase__ ,				sample=UpperCamelCase__ ,				generator=UpperCamelCase__ ,				prev_timestep=UpperCamelCase__ ,				).prev_sample

        A_  :	Tuple    						=  self.prior.post_process_latents(UpperCamelCase__  )

        A_  :	Any    						=  prior_latents

        A_  :	Union[str, Any]    						=  self.decoder_pipe(
            image=UpperCamelCase__ ,				height=UpperCamelCase__ ,				width=UpperCamelCase__ ,				num_inference_steps=UpperCamelCase__ ,				guidance_scale=UpperCamelCase__ ,				generator=UpperCamelCase__ ,				output_type=UpperCamelCase__ ,				return_dict=UpperCamelCase__ ,				num_images_per_prompt=UpperCamelCase__ ,				eta=UpperCamelCase__ ,				)
        return output






"""simple docstring"""




# Copyright 2023 The HuggingFace Team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_torch_available,
    is_vision_available,
)


lowerCamelCase_					:						Any      = {
    'configuration_tvlt': ['TVLT_PRETRAINED_CONFIG_ARCHIVE_MAP', 'TvltConfig'],
    'feature_extraction_tvlt': ['TvltFeatureExtractor'],
    'processing_tvlt': ['TvltProcessor'],
}

try:
   if not is_torch_available():
      raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
   pass
else:
   lowerCamelCase_					:						Union[str, Any]      = [
       'TVLT_PRETRAINED_MODEL_ARCHIVE_LIST',
       'TvltModel',
       'TvltForPreTraining',
       'TvltForAudioVisualClassification',
       'TvltPreTrainedModel',
   ]

try:
   if not is_vision_available():
      raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
   pass
else:
   lowerCamelCase_					:						List[Any]      = ['TvltImageProcessor']


if TYPE_CHECKING:
   from .configuration_tvlt import TVLT_PRETRAINED_CONFIG_ARCHIVE_MAP, TvltConfig
   from .processing_tvlt import TvltProcessor
   from .feature_extraction_tvlt import TvltFeatureExtractor

   try:
      if not is_torch_available():
         raise OptionalDependencyNotAvailable()
   except OptionalDependencyNotAvailable:
      pass
   else:
      from .modeling_tvlt import (
          TVLT_PRETRAINED_MODEL_ARCHIVE_LIST,
          TvltForAudioVisualClassification,
          TvltForPreTraining,
          TvltModel,
          TvltPreTrainedModel,
      )

   try:
      if not is_vision_available():
         raise OptionalDependencyNotAvailable()
   except OptionalDependencyNotAvailable:
      pass
   else:
      from .image_processing_tvlt import TvltImageProcessor


else:
   import sys

   lowerCamelCase_					:						Optional[int]      = _LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)