






"""simple docstring"""






from typing import List, Optional, Union

import numpy as np
import torch
import torchaudio.compliance.kaldi as ta_kaldi

from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import PaddingStrategy, TensorType, logging


SCREAMING_SNAKE_CASE    :    int        =							logging.get_logger(__name__)





class    snake_case_							(     A__							):



	"""simple docstring"""

	SCREAMING_SNAKE_CASE_:				List[Any]       =       ["""input_features""", """attention_mask"""]


	def __init__(   self	,       __a=80	,       __a=1_6000	,       __a=80	,       __a=0.0	,       __a=True	,       __a=True	,       __a=True	,       **__a	,       ):


					"""simple docstring"""

					super().__init__(feature_size=a__	,       sampling_rate=a__	,       padding_value=a__	,       **a__    )
					A__							=							num_mel_bins
					A__							=							do_ceptral_normalize
					A__							=							normalize_means
					A__							=							normalize_vars
					A__							=							True


	def    _UpperCAmelCase    (   self	,       __a	,       ):


					"""simple docstring"""

					A__							=							waveform * (2**15)  # Kaldi compliance: 16-bit signed integers
					A__							=							torch.from_numpy(a__    ).unsqueeze(0    )
					A__							=							ta_kaldi.fbank(a__	,       num_mel_bins=self.num_mel_bins	,       sample_frequency=self.sampling_rate    )
					return features.numpy()


	@staticmethod
	def    _UpperCAmelCase    (   __a	,       __a	,       __a = True	,       __a = True	,       __a = 0.0	,       ):


					"""simple docstring"""

					if normalize_means:
									A__							=							x[:input_length].mean(axis=0    )
									A__							=							np.subtract(a__	,       a__    )
					if normalize_vars:
									A__							=							x[:input_length].std(axis=0    )
									A__							=							np.divide(a__	,       a__    )

					if input_length < x.shape[0]:
									A__							=							padding_value

					# make sure array is in float32
					A__							=							x.astype(np.floataa    )

					return x


	def    _UpperCAmelCase    (   self	,       __a	,       __a = None    ):


					"""simple docstring"""

					A__							=							attention_mask.sum(-1    ) if attention_mask is not None else [x.shape[0] for x in input_features]
					return [
					    self.utterance_cmvn(a__	,       a__	,       self.normalize_means	,       self.normalize_vars	,       self.padding_value    )
					    for x, n in zip(a__	,       a__    )
					]






	def __call__(   self	,       __a	,       __a = False	,       __a = None	,       __a = False	,       __a = None	,       __a = None	,       __a = None	,       __a = None	,       **__a	,       ):


					"""simple docstring"""

					if sampling_rate is not None:
									if sampling_rate != self.sampling_rate:
													raise ValueError(
													    f'''The model corresponding to this feature extractor: {self} was trained using a sampling rate of'''
													    f''' {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled with'''
													    f''' {self.sampling_rate} and not {sampling_rate}.'''    )
					else:
									logger.warning(
									    'It is strongly recommended to pass the `sampling_rate` argument to this function. '
									    'Failing to do so can result in silent errors that might be hard to debug.'    )

					A__							=							isinstance(a__	,       np.ndarray    ) and len(raw_speech.shape    ) > 1
					if is_batched_numpy and len(raw_speech.shape    ) > 2:
									raise ValueError(f'''Only mono-channel audio is supported for input to {self}'''    )
					A__							=							is_batched_numpy or (
					    isinstance(a__	,       (list, tuple)    ) and (isinstance(raw_speech[0]	,       (np.ndarray, tuple, list)    ))
					)

					if is_batched:
									A__							=							[np.asarray(a__	,       dtype=np.floataa    ) for speech in raw_speech]
					elif not is_batched and not isinstance(a__	,       np.ndarray    ):
									A__							=							np.asarray(a__	,       dtype=np.floataa    )
					elif isinstance(a__	,       np.ndarray    ) and raw_speech.dtype is np.dtype(np.floataa    ):
									A__							=							raw_speech.astype(np.floataa    )

					# always return batch
					if not is_batched:
									A__							=							[raw_speech]

					# extract fbank features
					A__							=							[self._extract_fbank_features(a__    ) for waveform in raw_speech]

					# convert into correct format for padding
					A__							=							BatchFeature({'input_features': features}    )

					A__							=							self.pad(
					    a__	,       padding=a__	,       max_length=a__	,       truncation=a__	,       pad_to_multiple_of=a__	,       return_attention_mask=a__	,       **a__	,       )

					# make sure list is in array format
					A__							=							padded_inputs.get('input_features'    )
					if isinstance(input_features[0]	,       a__    ):
									A__							=							[np.asarray(a__	,       dtype=np.floataa    ) for feature in input_features]

					A__							=							padded_inputs.get('attention_mask'    )
					if attention_mask is not None:
									A__							=							[np.asarray(a__	,       dtype=np.intaa    ) for array in attention_mask]

					# Utterance-level cepstral mean and variance normalization
					if self.do_ceptral_normalize:
									A__							=							(
									    np.array(a__	,       dtype=np.intaa    )
									    if self._get_padding_strategies(a__	,       max_length=a__    ) is not PaddingStrategy.DO_NOT_PAD
									    else None
									)
									A__							=							self.normalize(
									    padded_inputs['input_features']	,       attention_mask=a__    )

					if return_tensors is not None:
									A__							=							padded_inputs.convert_to_tensors(a__    )

					return padded_inputs







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      _UpperCAmelCase      (							A__      ,   A__      ,   A__  ):
     UpperCamelCase__		     =					[r'''h\.\d+\.attn\.bias''', r'''h\.\d+\.attn\.masked_bias''']




     @register_to_config
     def __init__(					self			,    a__			,    a__			,    a__ = None			,    a__ = 5_0_2_5_7			,    a__ = 1_0_2_4			,    a__ = 7_6_8			,    a__ = 1_2			,    a__ = 1_2			,    a__ = None			,    a__ = "gelu_new"			,    a__ = 0.1			,    a__ = 0.1			,    a__ = 0.1			,    a__ = 1e-5			,    a__ = 0.0_2			,    a__ = True			,    a__ = True			,    a__ = False			,    a__ = False			,    ):
         super().__init__()

         A__					=       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.")

         A__					=       prefix_inner_dim
         A__					=       prefix_hidden_dim

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

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




     def 		snake_case_		(					self			,    a__			,    a__			,    a__ = None			,    a__ = None			,    ):
         A__					=       self.transformer.transformer.wte(a__)
         A__					=       self.encode_prefix(a__)
         A__					=       self.decode_prefix(a__)
         A__					=       torch.cat((prefix_embeds, embedding_text)			,    dim=1)

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




     def 		snake_case_		(					self			,    a__			,    a__):
         return torch.zeros(a__			,    self.prefix_length			,    dtype=torch.intaa			,    device=a__)




     def 		snake_case_		(					self			,    a__):
         return self.encode_prefix(a__)




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




     @torch.no_grad()
     def 		snake_case_		(					self			,    a__=None			,    a__=None			,    a__=None			,    a__ = 5			,    a__ = 6_7			,    a__ = 1.0			,    a__ = None			,    ):
         A__					=       eos_token_id
         A__					=       None
         A__					=       None
         A__					=       torch.ones(a__			,    device=a__			,    dtype=torch.int)
         A__					=       torch.zeros(a__			,    device=a__			,    dtype=torch.bool)

         if input_embeds is not None:
             A__					=       input_embeds
         else:
             A__					=       self.transformer.transformer.wte(a__)

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

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

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

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