

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

import torch
from transformers import (
    CLIPImageProcessor,
    CLIPTextModel,
    CLIPTokenizer,
    WhisperForConditionalGeneration,
    WhisperProcessor,
)

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    LMSDiscreteScheduler,
    PNDMScheduler,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion import StableDiffusionPipelineOutput
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker
from diffusers.utils import logging


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



class       __snake_case  (    lowerCAmelCase__       ):

  def __init__(					self						,		_A						,		_A						,		_A						,		_A						,		_A						,		_A						,		_A						,		_A						,		_A						,		):
      super().__init__()

      if safety_checker is None:
          logger.warning(
              f"""You have disabled the safety checker for {self.__class__} by passing `safety_checker=None`. Ensure"""
              ' that you abide to the conditions of the Stable Diffusion license and do not expose unfiltered'
              ' results in services or applications open to the public. Both the diffusers team and Hugging Face'
              ' strongly recommend to keep the safety filter enabled in all public facing circumstances, disabling'
              ' it only for use-cases that involve analyzing network behavior or auditing its results. For more'
              ' information, please have a look at https://github.com/huggingface/diffusers/pull/254 .')

      self.register_modules(
          speech_model=_A						,		speech_processor=_A						,		vae=_A						,		text_encoder=_A						,		tokenizer=_A						,		unet=_A						,		scheduler=_A						,		feature_extractor=_A						,		)

  def 						lowerCAmelCase__     (					self						,		_A = "auto"):
      if slice_size == "auto":
          SCREAMING_SNAKE_CASE_									=						self.unet.config.attention_head_dim // 2
      self.unet.set_attention_slice(_A)

  def 						lowerCAmelCase__     (					self):
      self.enable_attention_slicing(_A)


  @torch.no_grad()
  def __call__(					self						,		_A						,		_A=16000						,		_A = 512						,		_A = 512						,		_A = 50						,		_A = 7.5						,		_A = None						,		_A = 1						,		_A = 0.0						,		_A = None						,		_A = None						,		_A = "pil"						,		_A = True						,		_A = None						,		_A = 1						,		**_A						,		):
      SCREAMING_SNAKE_CASE_									=						self.speech_processor.feature_extractor(
          _A						,		return_tensors='pt'						,		sampling_rate=_A).input_features.to(self.device)
      SCREAMING_SNAKE_CASE_									=						self.speech_model.generate(_A						,		max_length=480000)

      SCREAMING_SNAKE_CASE_									=						self.speech_processor.tokenizer.batch_decode(_A						,		skip_special_tokens=_A						,		normalize=_A)[
          0
      ]

      if isinstance(_A						,		_A):
          SCREAMING_SNAKE_CASE_									=						1
      elif isinstance(_A						,		_A):
          SCREAMING_SNAKE_CASE_									=						len(_A)
      else:
          raise ValueError(f"""`prompt` has to be of type `str` or `list` but is {type(_A)}""")

      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(_A						,		_A) or callback_steps <= 0)
      ):
          raise ValueError(
              f"""`callback_steps` has to be a positive integer but is {callback_steps} of type"""
              f""" {type(_A)}.""")

      # get prompt text embeddings
      SCREAMING_SNAKE_CASE_									=						self.tokenizer(
          _A						,		padding='max_length'						,		max_length=self.tokenizer.model_max_length						,		return_tensors='pt'						,		)
      SCREAMING_SNAKE_CASE_									=						text_inputs.input_ids

      if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
          SCREAMING_SNAKE_CASE_									=						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}""")
          SCREAMING_SNAKE_CASE_									=						text_input_ids[:, : self.tokenizer.model_max_length]
      SCREAMING_SNAKE_CASE_									=						self.text_encoder(text_input_ids.to(self.device))[0]

      # duplicate text embeddings for each generation per prompt, using mps friendly method
      SCREAMING_SNAKE_CASE_							,			SCREAMING_SNAKE_CASE_							,			SCREAMING_SNAKE_CASE_									=						text_embeddings.shape
      SCREAMING_SNAKE_CASE_									=						text_embeddings.repeat(1						,		_A						,		1)
      SCREAMING_SNAKE_CASE_									=						text_embeddings.view(bs_embed * num_images_per_prompt						,		_A						,		-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.
      SCREAMING_SNAKE_CASE_									=						guidance_scale > 1.0
      # get unconditional embeddings for classifier free guidance
      if do_classifier_free_guidance:
          SCREAMING_SNAKE_CASE_									=						42
          if negative_prompt is None:
              SCREAMING_SNAKE_CASE_									=						[''] * batch_size
          elif type(_A) is not type(_A):
              raise TypeError(
                  f"""`negative_prompt` should be the same type to `prompt`, but got {type(_A)} !="""
                  f""" {type(_A)}.""")
          elif isinstance(_A						,		_A):
              SCREAMING_SNAKE_CASE_									=						[negative_prompt]
          elif batch_size != len(_A):
              raise ValueError(
                  f"""`negative_prompt`: {negative_prompt} has batch size {len(_A)}, but `prompt`:"""
                  f""" {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches"""
                  ' the batch size of `prompt`.')
          else:
              SCREAMING_SNAKE_CASE_									=						negative_prompt

          SCREAMING_SNAKE_CASE_									=						text_input_ids.shape[-1]
          SCREAMING_SNAKE_CASE_									=						self.tokenizer(
              _A						,		padding='max_length'						,		max_length=_A						,		truncation=_A						,		return_tensors='pt'						,		)
          SCREAMING_SNAKE_CASE_									=						self.text_encoder(uncond_input.input_ids.to(self.device))[0]

          # duplicate unconditional embeddings for each generation per prompt, using mps friendly method
          SCREAMING_SNAKE_CASE_									=						uncond_embeddings.shape[1]
          SCREAMING_SNAKE_CASE_									=						uncond_embeddings.repeat(1						,		_A						,		1)
          SCREAMING_SNAKE_CASE_									=						uncond_embeddings.view(batch_size * num_images_per_prompt						,		_A						,		-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
          SCREAMING_SNAKE_CASE_									=						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`.
      SCREAMING_SNAKE_CASE_									=						(batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
      SCREAMING_SNAKE_CASE_									=						text_embeddings.dtype
      if latents is None:
          if self.device.type == "mps":
              # randn does not exist on mps
              SCREAMING_SNAKE_CASE_									=						torch.randn(_A						,		generator=_A						,		device='cpu'						,		dtype=_A).to(
                  self.device)
          else:
              SCREAMING_SNAKE_CASE_									=						torch.randn(_A						,		generator=_A						,		device=self.device						,		dtype=_A)
      else:
          if latents.shape != latents_shape:
              raise ValueError(f"""Unexpected latents shape, got {latents.shape}, expected {latents_shape}""")
          SCREAMING_SNAKE_CASE_									=						latents.to(self.device)

      # set timesteps
      self.scheduler.set_timesteps(_A)

      # Some schedulers like PNDM have timesteps as arrays
      # It's more optimized to move all timesteps to correct device beforehand
      SCREAMING_SNAKE_CASE_									=						self.scheduler.timesteps.to(self.device)

      # scale the initial noise by the standard deviation required by the scheduler
      SCREAMING_SNAKE_CASE_									=						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]
      SCREAMING_SNAKE_CASE_									=						'eta' in set(inspect.signature(self.scheduler.step).parameters.keys())
      SCREAMING_SNAKE_CASE_									=						{}
      if accepts_eta:
          SCREAMING_SNAKE_CASE_									=						eta

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

          # predict the noise residual
          SCREAMING_SNAKE_CASE_									=						self.unet(_A						,		_A						,		encoder_hidden_states=_A).sample

          # perform guidance
          if do_classifier_free_guidance:
              SCREAMING_SNAKE_CASE_							,			SCREAMING_SNAKE_CASE_									=						noise_pred.chunk(2)
              SCREAMING_SNAKE_CASE_									=						noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

          # compute the previous noisy sample x_t -> x_t-1
          SCREAMING_SNAKE_CASE_									=						self.scheduler.step(_A						,		_A						,		_A						,		**_A).prev_sample

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

      SCREAMING_SNAKE_CASE_									=						1 / 0.1_8_2_1_5 * latents
      SCREAMING_SNAKE_CASE_									=						self.vae.decode(_A).sample

      SCREAMING_SNAKE_CASE_									=						(image / 2 + 0.5).clamp(0						,		1)

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

      if output_type == "pil":
          SCREAMING_SNAKE_CASE_									=						self.numpy_to_pil(_A)

      if not return_dict:
          return image

      return StableDiffusionPipelineOutput(images=_A						,		nsfw_content_detected=_A)





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


UpperCamelCase__  :      List[Any]			=    logging.get_logger(__name__)

UpperCamelCase__  :      List[str]			=    {
    "microsoft/biogpt": "https://huggingface.co/microsoft/biogpt/resolve/main/config.json",
    # See all BioGPT models at https://huggingface.co/models?filter=biogpt
}



class       __snake_case  (    lowerCAmelCase__       ):
  __lowerCAmelCase     :	Any						 =  'biogpt'

  def __init__(					self						,		_A=42384						,		_A=1024						,		_A=24						,		_A=16						,		_A=4096						,		_A="gelu"						,		_A=0.1						,		_A=0.1						,		_A=1024						,		_A=0.0_2						,		_A=1E-12						,		_A=True						,		_A=True						,		_A=0.0						,		_A=0.0						,		_A=1						,		_A=0						,		_A=2						,		**_A						,		):
      SCREAMING_SNAKE_CASE_									=						vocab_size
      SCREAMING_SNAKE_CASE_									=						max_position_embeddings
      SCREAMING_SNAKE_CASE_									=						hidden_size
      SCREAMING_SNAKE_CASE_									=						num_hidden_layers
      SCREAMING_SNAKE_CASE_									=						num_attention_heads
      SCREAMING_SNAKE_CASE_									=						intermediate_size
      SCREAMING_SNAKE_CASE_									=						hidden_act
      SCREAMING_SNAKE_CASE_									=						hidden_dropout_prob
      SCREAMING_SNAKE_CASE_									=						attention_probs_dropout_prob
      SCREAMING_SNAKE_CASE_									=						initializer_range
      SCREAMING_SNAKE_CASE_									=						layer_norm_eps
      SCREAMING_SNAKE_CASE_									=						scale_embedding
      SCREAMING_SNAKE_CASE_									=						use_cache
      SCREAMING_SNAKE_CASE_									=						layerdrop
      SCREAMING_SNAKE_CASE_									=						activation_dropout
      super().__init__(pad_token_id=_A						,		bos_token_id=_A						,		eos_token_id=_A						,		**_A)


