


"""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


__UpperCAmelCase    				=      logging.get_logger(__name__)  # pylint: disable=invalid-name





def     lowercase__	(  self     :  Union[str, Any]				,	lowerCamelCase     :  Any				,	lowerCamelCase     :  List[Any]				,	lowerCamelCase     :  Optional[int]				,	lowerCamelCase     :  Optional[Any]				)					->     Union[str, Any]:
  lowerCAmelCase__    :    Dict								=					image.to(device=lowerCamelCase				)
  lowerCAmelCase__    :    str								=					image  # take image as image_embeddings
  lowerCAmelCase__    :    Any								=					image_embeddings.unsqueeze(1				)

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

  if do_classifier_free_guidance:
    lowerCAmelCase__    :    Any								=					torch.zeros_like(lowerCamelCase				)

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

  return image_embeddings







class  lowercase_  ( a_						):







      def __init__(      self :							List[Any]		,  _lowercase :							PriorTransformer		,  _lowercase :							CLIPTokenizer		,  _lowercase :							CLIPTextModelWithProjection		,  _lowercase :							UnCLIPScheduler		,  _lowercase :							Optional[dict] = None		,  ):
        super().__init__()

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

        lowerCAmelCase__    :    Optional[int]								=					decoder_pipe_kwargs.get("torch_dtype"		,  _lowercase    ) or prior.dtype

        lowerCAmelCase__    :    Dict								=					StableDiffusionImageVariationPipeline.from_pretrained(
            "lambdalabs/sd-image-variations-diffusers"		,  **_lowercase    )

        # replace `_encode_image` method
        lowerCAmelCase__    :    str								=					types.MethodType(_encode_image		,  self.decoder_pipe    )

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







      def 				_lowerCAmelCase       (      self :							Dict		,  _lowercase :							int		,  _lowercase :							Union[str, Any]		,  _lowercase :							Any		,  _lowercase :							Union[str, Any]		,  _lowercase :							Optional[Union[CLIPTextModelOutput, Tuple]] = None		,  _lowercase :							Optional[torch.Tensor] = None		,  ):
        if text_model_output is None:
          lowerCAmelCase__    :    Optional[Any]								=					len(_lowercase    ) if isinstance(_lowercase		,  _lowercase    ) else 1
          # get prompt text embeddings
          lowerCAmelCase__    :    List[Any]								=					self.tokenizer(
              _lowercase		,  padding="max_length"		,  max_length=self.tokenizer.model_max_length		,  return_tensors="pt"		,  )
          lowerCAmelCase__    :    List[str]								=					text_inputs.input_ids
          lowerCAmelCase__    :    Union[str, Any]								=					text_inputs.attention_mask.bool().to(_lowercase    )

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

          lowerCAmelCase__    :    int								=					self.text_encoder(text_input_ids.to(_lowercase    )    )

          lowerCAmelCase__    :    Dict								=					text_encoder_output.text_embeds
          lowerCAmelCase__    :    Union[str, Any]								=					text_encoder_output.last_hidden_state

        else:
          lowerCAmelCase__    :    Any								=					text_model_output[0].shape[0]
          lowerCAmelCase__				,		lowerCAmelCase__    :    Union[str, Any]								=					text_model_output[0], text_model_output[1]
          lowerCAmelCase__    :    List[str]								=					text_attention_mask

        lowerCAmelCase__    :    Optional[int]								=					text_embeddings.repeat_interleave(_lowercase		,  dim=0    )
        lowerCAmelCase__    :    str								=					text_encoder_hidden_states.repeat_interleave(_lowercase		,  dim=0    )
        lowerCAmelCase__    :    List[str]								=					text_mask.repeat_interleave(_lowercase		,  dim=0    )

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

          lowerCAmelCase__    :    List[str]								=					self.tokenizer(
              _lowercase		,  padding="max_length"		,  max_length=self.tokenizer.model_max_length		,  truncation=_lowercase		,  return_tensors="pt"		,  )
          lowerCAmelCase__    :    Union[str, Any]								=					uncond_input.attention_mask.bool().to(_lowercase    )
          lowerCAmelCase__    :    Optional[int]								=					self.text_encoder(uncond_input.input_ids.to(_lowercase    )    )

          lowerCAmelCase__    :    Dict								=					uncond_embeddings_text_encoder_output.text_embeds
          lowerCAmelCase__    :    List[Any]								=					uncond_embeddings_text_encoder_output.last_hidden_state

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

          lowerCAmelCase__    :    List[Any]								=					uncond_embeddings.shape[1]
          lowerCAmelCase__    :    Optional[int]								=					uncond_embeddings.repeat(1		,  _lowercase    )
          lowerCAmelCase__    :    List[Any]								=					uncond_embeddings.view(batch_size * num_images_per_prompt		,  _lowercase    )

          lowerCAmelCase__    :    Dict								=					uncond_text_encoder_hidden_states.shape[1]
          lowerCAmelCase__    :    Any								=					uncond_text_encoder_hidden_states.repeat(1		,  _lowercase		,  1    )
          lowerCAmelCase__    :    Dict								=					uncond_text_encoder_hidden_states.view(
              batch_size * num_images_per_prompt		,  _lowercase		,  -1    )
          lowerCAmelCase__    :    Optional[Any]								=					uncond_text_mask.repeat_interleave(_lowercase		,  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
          lowerCAmelCase__    :    str								=					torch.cat([uncond_embeddings, text_embeddings]    )
          lowerCAmelCase__    :    List[Any]								=					torch.cat([uncond_text_encoder_hidden_states, text_encoder_hidden_states]    )

          lowerCAmelCase__    :    str								=					torch.cat([uncond_text_mask, text_mask]    )

        return text_embeddings, text_encoder_hidden_states, text_mask







      @property
      def 				_lowerCAmelCase       (      self :							List[str]    ):
        if self.device != torch.device("meta"    ) or not hasattr(self.prior		,  "_hf_hook"    ):
          return self.device
        for module in self.prior.modules():
          if (
              hasattr(_lowercase		,  "_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 :							Optional[int]		,  _lowercase :							Union[str, Any]		,  _lowercase :							List[str]		,  _lowercase :							Optional[Any]		,  _lowercase :							Dict		,  _lowercase :							int		,  _lowercase :							Any    ):
        if latents is None:
          lowerCAmelCase__    :    Optional[Any]								=					randn_tensor(_lowercase		,  generator=_lowercase		,  device=_lowercase		,  dtype=_lowercase    )
        else:
          if latents.shape != shape:
            raise ValueError(f"Unexpected latents shape, got {latents.shape}, expected {shape}"    )
          lowerCAmelCase__    :    Any								=					latents.to(_lowercase    )

        lowerCAmelCase__    :    Union[str, Any]								=					latents * scheduler.init_noise_sigma
        return latents







      def 				_lowerCAmelCase       (      self :							Union[str, Any]		,  _lowercase :							Optional[Union[str, torch.device]] = None    ):
        self.decoder_pipe.to(_lowercase    )
        super().to(_lowercase    )







      @torch.no_grad()
      def __call__(      self :							Dict		,  _lowercase :							Optional[Union[str, List[str]]] = None		,  _lowercase :							Optional[int] = None		,  _lowercase :							Optional[int] = None		,  _lowercase :							int = 1		,  _lowercase :							int = 2_5		,  _lowercase :							Optional[torch.Generator] = None		,  _lowercase :							Optional[torch.FloatTensor] = None		,  _lowercase :							Optional[Union[CLIPTextModelOutput, Tuple]] = None		,  _lowercase :							Optional[torch.Tensor] = None		,  _lowercase :							float = 4.0		,  _lowercase :							float = 8.0		,  _lowercase :							int = 5_0		,  _lowercase :							Optional[int] = 1		,  _lowercase :							float = 0.0		,  _lowercase :							Optional[str] = "pil"		,  _lowercase :							bool = True		,  ):
        if prompt is not None:
          if isinstance(_lowercase		,  _lowercase    ):
            lowerCAmelCase__    :    Optional[int]								=					1
          elif isinstance(_lowercase		,  _lowercase    ):
            lowerCAmelCase__    :    List[Any]								=					len(_lowercase    )
          else:
            raise ValueError(f"`prompt` has to be of type `str` or `list` but is {type(_lowercase    )}"    )
        else:
          lowerCAmelCase__    :    Union[str, Any]								=					text_model_output[0].shape[0]

        lowerCAmelCase__    :    List[Any]								=					self._execution_device

        lowerCAmelCase__    :    Optional[int]								=					batch_size * num_images_per_prompt

        lowerCAmelCase__    :    Dict								=					prior_guidance_scale > 1.0 or decoder_guidance_scale > 1.0

        lowerCAmelCase__				,		lowerCAmelCase__				,		lowerCAmelCase__    :    Dict								=					self._encode_prompt(
            _lowercase		,  _lowercase		,  _lowercase		,  _lowercase		,  _lowercase		,  _lowercase    )

        # prior

        self.prior_scheduler.set_timesteps(_lowercase		,  device=_lowercase    )
        lowerCAmelCase__    :    str								=					self.prior_scheduler.timesteps

        lowerCAmelCase__    :    List[str]								=					self.prior.config.embedding_dim

        lowerCAmelCase__    :    Union[str, Any]								=					self.prepare_latents(
            (batch_size, embedding_dim)		,  text_embeddings.dtype		,  _lowercase		,  _lowercase		,  _lowercase		,  self.prior_scheduler		,  )

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

          lowerCAmelCase__    :    Dict								=					self.prior(
              _lowercase		,  timestep=_lowercase		,  proj_embedding=_lowercase		,  encoder_hidden_states=_lowercase		,  attention_mask=_lowercase		,  ).predicted_image_embedding

          if do_classifier_free_guidance:
            lowerCAmelCase__				,		lowerCAmelCase__    :    Any								=					predicted_image_embedding.chunk(2    )
            lowerCAmelCase__    :    Tuple								=					predicted_image_embedding_uncond + prior_guidance_scale * (
                predicted_image_embedding_text - predicted_image_embedding_uncond
            )

          if i + 1 == prior_timesteps_tensor.shape[0]:
            lowerCAmelCase__    :    int								=					None
          else:
            lowerCAmelCase__    :    str								=					prior_timesteps_tensor[i + 1]

          lowerCAmelCase__    :    Tuple								=					self.prior_scheduler.step(
              _lowercase		,  timestep=_lowercase		,  sample=_lowercase		,  generator=_lowercase		,  prev_timestep=_lowercase		,  ).prev_sample

        lowerCAmelCase__    :    Optional[Any]								=					self.prior.post_process_latents(_lowercase    )

        lowerCAmelCase__    :    str								=					prior_latents

        lowerCAmelCase__    :    str								=					self.decoder_pipe(
            image=_lowercase		,  height=_lowercase		,  width=_lowercase		,  num_inference_steps=_lowercase		,  guidance_scale=_lowercase		,  generator=_lowercase		,  output_type=_lowercase		,  return_dict=_lowercase		,  num_images_per_prompt=_lowercase		,  eta=_lowercase		,  )
        return output




"""simple docstring"""





import functools
import operator

from ...configuration_utils import PretrainedConfig
from ...utils import logging


__UpperCAmelCase    				=      logging.get_logger(__name__)

__UpperCAmelCase    				=      {
    "facebook/hubert-base-ls960": "https://huggingface.co/facebook/hubert-base-ls960/resolve/main/config.json",
    # See all Hubert models at https://huggingface.co/models?filter=hubert
}

class  lowercase_  ( a_						):
      __magic_name__     :							List[Any]          =				"""hubert"""







      def __init__(      self :							List[Any]		,  _lowercase :							Optional[int]=3_2		,  _lowercase :							Tuple=7_6_8		,  _lowercase :							List[Any]=1_2		,  _lowercase :							Dict=1_2		,  _lowercase :							Tuple=3_0_7_2		,  _lowercase :							str="gelu"		,  _lowercase :							Optional[int]=0.1		,  _lowercase :							int=0.1		,  _lowercase :							Tuple=0.1		,  _lowercase :							str=True		,  _lowercase :							Optional[Any]=0.0		,  _lowercase :							List[Any]=0.1		,  _lowercase :							Union[str, Any]=0.1		,  _lowercase :							Union[str, Any]=0.02		,  _lowercase :							List[Any]=1e-5		,  _lowercase :							List[Any]="group"		,  _lowercase :							Tuple="gelu"		,  _lowercase :							List[str]=(5_1_2, 5_1_2, 5_1_2, 5_1_2, 5_1_2, 5_1_2, 5_1_2)		,  _lowercase :							List[Any]=(5, 2, 2, 2, 2, 2, 2)		,  _lowercase :							Tuple=(1_0, 3, 3, 3, 3, 2, 2)		,  _lowercase :							Any=False		,  _lowercase :							str=1_2_8		,  _lowercase :							int=1_6		,  _lowercase :							str=False		,  _lowercase :							Union[str, Any]=True		,  _lowercase :							Optional[Any]=0.05		,  _lowercase :							List[Any]=1_0		,  _lowercase :							Dict=2		,  _lowercase :							Optional[int]=0.0		,  _lowercase :							List[str]=1_0		,  _lowercase :							List[str]=0		,  _lowercase :							Optional[Any]="sum"		,  _lowercase :							int=False		,  _lowercase :							Any=False		,  _lowercase :							Optional[Any]=2_5_6		,  _lowercase :							int=0		,  _lowercase :							Optional[int]=1		,  _lowercase :							List[Any]=2		,  **_lowercase :							Tuple		,  ):
        super().__init__(**_lowercase		,  pad_token_id=_lowercase		,  bos_token_id=_lowercase		,  eos_token_id=_lowercase    )
        lowerCAmelCase__    :    Optional[Any]								=					hidden_size
        lowerCAmelCase__    :    int								=					feat_extract_norm
        lowerCAmelCase__    :    Optional[Any]								=					feat_extract_activation
        lowerCAmelCase__    :    Optional[Any]								=					list(_lowercase    )
        lowerCAmelCase__    :    Union[str, Any]								=					list(_lowercase    )
        lowerCAmelCase__    :    Tuple								=					list(_lowercase    )
        lowerCAmelCase__    :    str								=					conv_bias
        lowerCAmelCase__    :    Optional[Any]								=					num_conv_pos_embeddings
        lowerCAmelCase__    :    List[str]								=					num_conv_pos_embedding_groups
        lowerCAmelCase__    :    List[Any]								=					len(self.conv_dim    )
        lowerCAmelCase__    :    Union[str, Any]								=					num_hidden_layers
        lowerCAmelCase__    :    str								=					intermediate_size
        lowerCAmelCase__    :    List[Any]								=					hidden_act
        lowerCAmelCase__    :    Optional[int]								=					num_attention_heads
        lowerCAmelCase__    :    Optional[int]								=					hidden_dropout
        lowerCAmelCase__    :    Dict								=					attention_dropout
        lowerCAmelCase__    :    Dict								=					activation_dropout
        lowerCAmelCase__    :    Optional[Any]								=					feat_proj_layer_norm
        lowerCAmelCase__    :    Union[str, Any]								=					feat_proj_dropout
        lowerCAmelCase__    :    Optional[Any]								=					final_dropout
        lowerCAmelCase__    :    Dict								=					layerdrop
        lowerCAmelCase__    :    Optional[int]								=					layer_norm_eps
        lowerCAmelCase__    :    str								=					initializer_range
        lowerCAmelCase__    :    int								=					vocab_size
        lowerCAmelCase__    :    int								=					do_stable_layer_norm
        lowerCAmelCase__    :    Union[str, Any]								=					use_weighted_layer_sum
        lowerCAmelCase__    :    Tuple								=					classifier_proj_size

        if (
            (len(self.conv_stride    ) != self.num_feat_extract_layers)
            or (len(self.conv_kernel    ) != self.num_feat_extract_layers)
            or (len(self.conv_dim    ) != self.num_feat_extract_layers)
        ):
          raise ValueError(
              "Configuration for convolutional layers is incorrect. It is required that `len(config.conv_dim)` =="
              " `len(config.conv_stride)` == `len(config.conv_kernel)`, but is `len(config.conv_dim) ="
              f" {len(self.conv_dim    )}`, `len(config.conv_stride) = {len(self.conv_stride    )}`,"
              f" `len(config.conv_kernel) = {len(self.conv_kernel    )}`."    )

        # fine-tuning config parameters for SpecAugment: https://arxiv.org/abs/1904.08779
        lowerCAmelCase__    :    int								=					apply_spec_augment
        lowerCAmelCase__    :    Tuple								=					mask_time_prob
        lowerCAmelCase__    :    int								=					mask_time_length
        lowerCAmelCase__    :    Dict								=					mask_time_min_masks
        lowerCAmelCase__    :    Union[str, Any]								=					mask_feature_prob
        lowerCAmelCase__    :    Dict								=					mask_feature_length
        lowerCAmelCase__    :    Tuple								=					mask_feature_min_masks

        # ctc loss
        lowerCAmelCase__    :    Tuple								=					ctc_loss_reduction
        lowerCAmelCase__    :    List[str]								=					ctc_zero_infinity







      @property
      def 				_lowerCAmelCase       (      self :							Dict    ):
        return functools.reduce(operator.mul		,  self.conv_stride		,  1    )
