



"""simple docstring"""
from typing import Dict, List, Optional, Union

import numpy as np

from .feature_extraction_utils import BatchFeature, FeatureExtractionMixin
from .utils import PaddingStrategy, TensorType, is_tf_tensor, is_torch_tensor, logging, to_numpy


UpperCAmelCase						:		Optional[Any]			= logging.get_logger(__name__)


class    SCREAMING_SNAKE_CASE__ (							__UpperCAmelCase   ):

    def __init__(				self			:				Optional[Any] ,  lowerCAmelCase_			:				int ,  lowerCAmelCase_			:				int ,  lowerCAmelCase_			:				float ,  **lowerCAmelCase_			:				Optional[int]):







      """simple docstring"""







      lowercase_								= feature_size
      lowercase_								= sampling_rate
      lowercase_								= padding_value

      lowercase_								= kwargs.pop("""padding_side""" ,  """right""")
      lowercase_								= kwargs.pop("""return_attention_mask""" ,  lowerCAmelCase_)

      super().__init__(**lowerCAmelCase_)

    def 					_UpperCAmelCase	(				self			:				int ,  lowerCAmelCase_			:				Union[
            BatchFeature,
            List[BatchFeature],
            Dict[str, BatchFeature],
            Dict[str, List[BatchFeature]],
            List[Dict[str, BatchFeature]],
        ] ,  lowerCAmelCase_			:				Union[bool, str, PaddingStrategy] = True ,  lowerCAmelCase_			:				Optional[int] = None ,  lowerCAmelCase_			:				bool = False ,  lowerCAmelCase_			:				Optional[int] = None ,  lowerCAmelCase_			:				Optional[bool] = None ,  lowerCAmelCase_			:				Optional[Union[str, TensorType]] = None ,  ):







      """simple docstring"""







      if isinstance(lowerCAmelCase_ ,  (list, tuple)) and isinstance(processed_features[0] ,  (dict, BatchFeature)):
        lowercase_								= {
            key: [example[key] for example in processed_features] for key in processed_features[0].keys()
        }

      # The model's main input name, usually `input_values`, has be passed for padding
      if self.model_input_names[0] not in processed_features:
        raise ValueError(
            """You should supply an instance of `transformers.BatchFeature` or list of `transformers.BatchFeature`"""
            F''' to this method that includes {self.model_input_names[0]}, but you provided'''
            F''' {list(processed_features.keys())}''')

      lowercase_								= processed_features[self.model_input_names[0]]
      lowercase_								= (
          return_attention_mask if return_attention_mask is not None else self.return_attention_mask
      )

      if len(lowerCAmelCase_) == 0:
        if return_attention_mask:
          lowercase_								= []
        return processed_features

      # If we have PyTorch/TF tensors or lists as inputs, we cast them as Numpy arrays
      # and rebuild them afterwards if no return_tensors is specified
      # Note that we lose the specific device the tensor may be on for PyTorch

      lowercase_								= required_input[0]
      if isinstance(lowerCAmelCase_ ,  (list, tuple)):
        # first_element might be an empty list/tuple in some edge cases so we grab the first non empty element.
        lowercase_								= 0
        while len(required_input[index]) == 0:
          index += 1
        if index < len(lowerCAmelCase_):
          lowercase_								= required_input[index][0]

      if return_tensors is None:
        if is_tf_tensor(lowerCAmelCase_):
          lowercase_								= """tf"""
        elif is_torch_tensor(lowerCAmelCase_):
          lowercase_								= """pt"""
        elif isinstance(lowerCAmelCase_ ,  (int, float, list, tuple, np.ndarray)):
          lowercase_								= """np"""
        else:
          raise ValueError(
              F'''type of {first_element} unknown: {type(lowerCAmelCase_)}. '''
              """Should be one of a python, numpy, pytorch or tensorflow object.""")

      for key, value in processed_features.items():
        if isinstance(value[0] ,  (int, float)):
          lowercase_								= to_numpy(lowerCAmelCase_)
        else:
          lowercase_								= [to_numpy(lowerCAmelCase_) for v in value]

        # Convert padding_strategy in PaddingStrategy
      lowercase_								= self._get_padding_strategies(padding=lowerCAmelCase_ ,  max_length=lowerCAmelCase_)

      lowercase_								= processed_features[self.model_input_names[0]]

      lowercase_								= len(lowerCAmelCase_)
      if not all(len(lowerCAmelCase_) == batch_size for v in processed_features.values()):
        raise ValueError("""Some items in the output dictionary have a different batch size than others.""")

      lowercase_								= []
      for i in range(lowerCAmelCase_):
        lowercase_								= {k: v[i] for k, v in processed_features.items()}
        # truncation
        lowercase_								= self._truncate(
            lowerCAmelCase_ ,  max_length=lowerCAmelCase_ ,  pad_to_multiple_of=lowerCAmelCase_ ,  truncation=lowerCAmelCase_ ,  )
        truncated_inputs.append(lowerCAmelCase_)

      if padding_strategy == PaddingStrategy.LONGEST:
        # make sure that `max_length` cannot be longer than the longest truncated length
        lowercase_								= max(len(input_slice[self.model_input_names[0]]) for input_slice in truncated_inputs)
        lowercase_								= PaddingStrategy.MAX_LENGTH

      lowercase_								= {}
      for i in range(lowerCAmelCase_):
        # padding
        lowercase_								= self._pad(
            truncated_inputs[i] ,  max_length=lowerCAmelCase_ ,  padding_strategy=lowerCAmelCase_ ,  pad_to_multiple_of=lowerCAmelCase_ ,  return_attention_mask=lowerCAmelCase_ ,  )

        for key, value in outputs.items():
          if key not in batch_outputs:
            lowercase_								= []
          if value.dtype is np.dtype(np.floataa):
            lowercase_								= value.astype(np.floataa)
          batch_outputs[key].append(lowerCAmelCase_)

      return BatchFeature(lowerCAmelCase_ ,  tensor_type=lowerCAmelCase_)

    def 					_UpperCAmelCase	(				self			:				Dict ,  lowerCAmelCase_			:				Union[Dict[str, np.ndarray], BatchFeature] ,  lowerCAmelCase_			:				Optional[int] = None ,  lowerCAmelCase_			:				PaddingStrategy = PaddingStrategy.DO_NOT_PAD ,  lowerCAmelCase_			:				Optional[int] = None ,  lowerCAmelCase_			:				Optional[bool] = None ,  ):







      """simple docstring"""







      lowercase_								= processed_features[self.model_input_names[0]]

      if padding_strategy == PaddingStrategy.LONGEST:
        lowercase_								= len(lowerCAmelCase_)

      if max_length is not None and pad_to_multiple_of is not None and (max_length % pad_to_multiple_of != 0):
        lowercase_								= ((max_length // pad_to_multiple_of) + 1) * pad_to_multiple_of

      lowercase_								= padding_strategy != PaddingStrategy.DO_NOT_PAD and len(lowerCAmelCase_) < max_length

      if return_attention_mask and "attention_mask" not in processed_features:
        lowercase_								= np.ones(len(lowerCAmelCase_) ,  dtype=np.intaa)

      if needs_to_be_padded:
        lowercase_								= max_length - len(lowerCAmelCase_)
        if self.padding_side == "right":
          if return_attention_mask:
            lowercase_								= np.pad(
                processed_features["""attention_mask"""] ,  (0, difference))
          lowercase_								= ((0, difference), (0, 0)) if self.feature_size > 1 else (0, difference)
          lowercase_								= np.pad(
              lowerCAmelCase_ ,  lowerCAmelCase_ ,  """constant""" ,  constant_values=self.padding_value)
        elif self.padding_side == "left":
          if return_attention_mask:
            lowercase_								= np.pad(
                processed_features["""attention_mask"""] ,  (difference, 0))
          lowercase_								= ((difference, 0), (0, 0)) if self.feature_size > 1 else (difference, 0)
          lowercase_								= np.pad(
              lowerCAmelCase_ ,  lowerCAmelCase_ ,  """constant""" ,  constant_values=self.padding_value)
        else:
          raise ValueError("""Invalid padding strategy:""" + str(self.padding_side))

      return processed_features

    def 					_UpperCAmelCase	(				self			:				int ,  lowerCAmelCase_			:				Union[Dict[str, np.ndarray], BatchFeature] ,  lowerCAmelCase_			:				Optional[int] = None ,  lowerCAmelCase_			:				Optional[int] = None ,  lowerCAmelCase_			:				Optional[bool] = None ,  ):







      """simple docstring"""







      if not truncation:
        return processed_features
      elif truncation and max_length is None:
        raise ValueError("""When setting ``truncation=True``, make sure that ``max_length`` is defined.""")

      lowercase_								= processed_features[self.model_input_names[0]]

      # find `max_length` that fits `pad_to_multiple_of`
      if max_length is not None and pad_to_multiple_of is not None and (max_length % pad_to_multiple_of != 0):
        lowercase_								= ((max_length // pad_to_multiple_of) + 1) * pad_to_multiple_of

      lowercase_								= len(lowerCAmelCase_) > max_length

      if needs_to_be_truncated:
        lowercase_								= processed_features[self.model_input_names[0]][:max_length]
        if "attention_mask" in processed_features:
          lowercase_								= processed_features["""attention_mask"""][:max_length]

      return processed_features




    def 					_UpperCAmelCase	(				self			:				Union[str, Any] ,  lowerCAmelCase_			:				Optional[Any]=False ,  lowerCAmelCase_			:				List[Any]=None):







      """simple docstring"""







      if padding is not False:
        if padding is True:
          lowercase_								= PaddingStrategy.LONGEST  # Default to pad to the longest sequence in the batch
        elif not isinstance(lowerCAmelCase_ ,  lowerCAmelCase_):
          lowercase_								= PaddingStrategy(lowerCAmelCase_)
        elif isinstance(lowerCAmelCase_ ,  lowerCAmelCase_):
          lowercase_								= padding
      else:
        lowercase_								= PaddingStrategy.DO_NOT_PAD

      # Set max length if needed
      if max_length is None:
        if padding_strategy == PaddingStrategy.MAX_LENGTH:
          raise ValueError(
              F'''When setting ``padding={PaddingStrategy.MAX_LENGTH}``, make sure that max_length is defined''')

        # Test if we have a padding value
      if padding_strategy != PaddingStrategy.DO_NOT_PAD and (self.padding_value is None):
        raise ValueError(
            """Asking to pad but the feature_extractor does not have a padding value. Please select a value to use"""
            """ as `padding_value`. For example: `feature_extractor.padding_value = 0.0`.""")

      return padding_strategy






"""simple docstring"""
from typing import Optional

import numpy as np
import torch
from torch import nn
from transformers import GPTaConfig, GPTaLMHeadModel
from transformers.modeling_utils import ModuleUtilsMixin

from ...configuration_utils import ConfigMixin, register_to_config
from ...models import ModelMixin


class    SCREAMING_SNAKE_CASE__ (							__UpperCAmelCase      ,  __UpperCAmelCase      ,  __UpperCAmelCase   ):
    lowercase__      =   [r"h\.\d+\.attn\.bias", r"h\.\d+\.attn\.masked_bias"]

    @register_to_config
    def __init__(				self			:				int ,  lowerCAmelCase_			:				int ,  lowerCAmelCase_			:				int ,  lowerCAmelCase_			:				Optional[int] = None ,  lowerCAmelCase_			:				int = 5_0_2_5_7 ,  lowerCAmelCase_			:				int = 1_0_2_4 ,  lowerCAmelCase_			:				int = 7_6_8 ,  lowerCAmelCase_			:				int = 1_2 ,  lowerCAmelCase_			:				int = 1_2 ,  lowerCAmelCase_			:				Optional[int] = None ,  lowerCAmelCase_			:				str = "gelu_new" ,  lowerCAmelCase_			:				float = 0.1 ,  lowerCAmelCase_			:				float = 0.1 ,  lowerCAmelCase_			:				float = 0.1 ,  lowerCAmelCase_			:				float = 1E-5 ,  lowerCAmelCase_			:				float = 0.02 ,  lowerCAmelCase_			:				bool = True ,  lowerCAmelCase_			:				bool = True ,  lowerCAmelCase_			:				bool = False ,  lowerCAmelCase_			:				bool = False ,  ):







      """simple docstring"""







      super().__init__()

      lowercase_								= prefix_length

      if prefix_inner_dim != n_embd and prefix_hidden_dim is None:
        raise ValueError(
            F'''`prefix_hidden_dim` cannot be `None` when `prefix_inner_dim`: {prefix_hidden_dim} and'''
            F''' `n_embd`: {n_embd} are not equal.''')

      lowercase_								= prefix_inner_dim
      lowercase_								= prefix_hidden_dim

      lowercase_								= (
          nn.Linear(self.prefix_inner_dim ,  self.prefix_hidden_dim)
          if self.prefix_hidden_dim is not None
          else nn.Identity()
      )
      lowercase_								= (
          nn.Linear(self.prefix_hidden_dim ,  lowerCAmelCase_) if self.prefix_hidden_dim is not None else nn.Identity()
      )

      lowercase_								= GPTaConfig(
          vocab_size=lowerCAmelCase_ ,  n_positions=lowerCAmelCase_ ,  n_embd=lowerCAmelCase_ ,  n_layer=lowerCAmelCase_ ,  n_head=lowerCAmelCase_ ,  n_inner=lowerCAmelCase_ ,  activation_function=lowerCAmelCase_ ,  resid_pdrop=lowerCAmelCase_ ,  embd_pdrop=lowerCAmelCase_ ,  attn_pdrop=lowerCAmelCase_ ,  layer_norm_epsilon=lowerCAmelCase_ ,  initializer_range=lowerCAmelCase_ ,  scale_attn_weights=lowerCAmelCase_ ,  use_cache=lowerCAmelCase_ ,  scale_attn_by_inverse_layer_idx=lowerCAmelCase_ ,  reorder_and_upcast_attn=lowerCAmelCase_ ,  )
      lowercase_								= GPTaLMHeadModel(lowerCAmelCase_)

    def 					_UpperCAmelCase	(				self			:				int ,  lowerCAmelCase_			:				torch.Tensor ,  lowerCAmelCase_			:				torch.Tensor ,  lowerCAmelCase_			:				Optional[torch.Tensor] = None ,  lowerCAmelCase_			:				Optional[torch.Tensor] = None ,  ):







      """simple docstring"""







      lowercase_								= self.transformer.transformer.wte(lowerCAmelCase_)
      lowercase_								= self.encode_prefix(lowerCAmelCase_)
      lowercase_								= self.decode_prefix(lowerCAmelCase_)
      lowercase_								= torch.cat((prefix_embeds, embedding_text) ,  dim=1)

      if labels is not None:
        lowercase_								= self.get_dummy_token(input_ids.shape[0] ,  input_ids.device)
        lowercase_								= torch.cat((dummy_token, input_ids) ,  dim=1)
      lowercase_								= self.transformer(inputs_embeds=lowerCAmelCase_ ,  labels=lowerCAmelCase_ ,  attention_mask=lowerCAmelCase_)
      if self.prefix_hidden_dim is not None:
        return out, hidden
      else:
        return out

    def 					_UpperCAmelCase	(				self			:				Optional[int] ,  lowerCAmelCase_			:				int ,  lowerCAmelCase_			:				torch.device):







      """simple docstring"""







      return torch.zeros(lowerCAmelCase_ ,  self.prefix_length ,  dtype=torch.intaa ,  device=lowerCAmelCase_)

    def 					_UpperCAmelCase	(				self			:				List[str] ,  lowerCAmelCase_			:				List[Any]):







      """simple docstring"""







      return self.encode_prefix(lowerCAmelCase_)

    @torch.no_grad()
    def 					_UpperCAmelCase	(				self			:				Tuple ,  lowerCAmelCase_			:				str ,  lowerCAmelCase_			:				Dict ,  lowerCAmelCase_			:				Union[str, Any]):







      """simple docstring"""







      lowercase_								= torch.split(lowerCAmelCase_ ,  1 ,  dim=0)
      lowercase_								= []
      lowercase_								= []
      for feature in features:
        lowercase_								= self.decode_prefix(feature.to(lowerCAmelCase_))  # back to the clip feature
        # Only support beam search for now
        lowercase_					,    lowercase_								= self.generate_beam(
            input_embeds=lowerCAmelCase_ ,  device=lowerCAmelCase_ ,  eos_token_id=lowerCAmelCase_)
        generated_tokens.append(output_tokens[0])
        generated_seq_lengths.append(seq_lengths[0])
      lowercase_								= torch.stack(lowerCAmelCase_)
      lowercase_								= torch.stack(lowerCAmelCase_)
      return generated_tokens, generated_seq_lengths




    @torch.no_grad()
    def 					_UpperCAmelCase	(				self			:				int ,  lowerCAmelCase_			:				List[str]=None ,  lowerCAmelCase_			:				Tuple=None ,  lowerCAmelCase_			:				str=None ,  lowerCAmelCase_			:				int = 5 ,  lowerCAmelCase_			:				int = 6_7 ,  lowerCAmelCase_			:				float = 1.0 ,  lowerCAmelCase_			:				Optional[int] = None ,  ):







      """simple docstring"""







      lowercase_								= eos_token_id
      lowercase_								= None
      lowercase_								= None
      lowercase_								= torch.ones(lowerCAmelCase_ ,  device=lowerCAmelCase_ ,  dtype=torch.int)
      lowercase_								= torch.zeros(lowerCAmelCase_ ,  device=lowerCAmelCase_ ,  dtype=torch.bool)

      if input_embeds is not None:
        lowercase_								= input_embeds
      else:
        lowercase_								= self.transformer.transformer.wte(lowerCAmelCase_)

      for i in range(lowerCAmelCase_):
        lowercase_								= self.transformer(inputs_embeds=lowerCAmelCase_)
        lowercase_								= outputs.logits
        lowercase_								= logits[:, -1, :] / (temperature if temperature > 0 else 1.0)
        lowercase_								= logits.softmax(-1).log()

        if scores is None:
          lowercase_					,    lowercase_								= logits.topk(lowerCAmelCase_ ,  -1)
          lowercase_								= generated.expand(lowerCAmelCase_ ,  *generated.shape[1:])
          lowercase_					,    lowercase_								= next_tokens.permute(1 ,  0), scores.squeeze(0)
          if tokens is None:
            lowercase_								= next_tokens
          else:
            lowercase_								= tokens.expand(lowerCAmelCase_ ,  *tokens.shape[1:])
            lowercase_								= torch.cat((tokens, next_tokens) ,  dim=1)
        else:
          lowercase_								= -float(np.inf)
          lowercase_								= 0
          lowercase_								= scores[:, None] + logits
          seq_lengths[~is_stopped] += 1
          lowercase_								= scores_sum / seq_lengths[:, None]
          lowercase_					,    lowercase_								= scores_sum_average.view(-1).topk(lowerCAmelCase_ ,  -1)
          lowercase_								= next_tokens // scores_sum.shape[1]
          lowercase_								= seq_lengths[next_tokens_source]
          lowercase_								= next_tokens % scores_sum.shape[1]
          lowercase_								= next_tokens.unsqueeze(1)
          lowercase_								= tokens[next_tokens_source]
          lowercase_								= torch.cat((tokens, next_tokens) ,  dim=1)
          lowercase_								= generated[next_tokens_source]
          lowercase_								= scores_sum_average * seq_lengths
          lowercase_								= is_stopped[next_tokens_source]

        lowercase_								= self.transformer.transformer.wte(next_tokens.squeeze()).view(generated.shape[0] ,  1 ,  -1)
        lowercase_								= torch.cat((generated, next_token_embed) ,  dim=1)
        lowercase_								= is_stopped + next_tokens.eq(lowerCAmelCase_).squeeze()
        if is_stopped.all():
          break

      lowercase_								= scores / seq_lengths
      lowercase_								= scores.argsort(descending=lowerCAmelCase_)
      # tokens tensors are already padded to max_seq_length
      lowercase_								= [tokens[i] for i in order]
      lowercase_								= torch.stack(lowerCAmelCase_ ,  dim=0)
      lowercase_								= torch.tensor([seq_lengths[i] for i in order] ,  dtype=seq_lengths.dtype)
      return output_texts, seq_lengths

