






"""simple docstring"""


import torch
from torch import nn


class 					snake_case		(							nn.Module						):


   '''simple docstring'''

   def __init__(	self :     List[Any],       _lowerCamelCase :     Tuple,       _lowerCamelCase :     int,       _lowerCamelCase :     List[Any],       _lowerCamelCase :     Any,       _lowerCamelCase :     Any=1,       _lowerCamelCase :     Tuple=False  ):

     '''simple docstring'''
     super().__init__()

     __A									= n_token
     __A									= d_embed
     __A									= d_proj

     __A									= cutoffs + [n_token]
     __A									= [0] + self.cutoffs
     __A									= div_val

     __A									= self.cutoffs[0]
     __A									= len(self.cutoffs  ) - 1
     __A									= self.shortlist_size + self.n_clusters

     if self.n_clusters > 0:
       __A									= nn.Parameter(torch.zeros(self.n_clusters,       self.d_embed  )  )
       __A									= nn.Parameter(torch.zeros(self.n_clusters  )  )

     __A									= nn.ModuleList()
     __A									= nn.ParameterList()

     if div_val == 1:
       for i in range(len(self.cutoffs  )  ):
         if d_proj != d_embed:
           self.out_projs.append(nn.Parameter(torch.FloatTensor(_lowerCamelCase,       _lowerCamelCase  )  )  )
         else:
           self.out_projs.append(_lowerCamelCase  )

       self.out_layers.append(nn.Linear(_lowerCamelCase,       _lowerCamelCase  )  )
     else:
       for i in range(len(self.cutoffs  )  ):
         __A			,    __A									= self.cutoff_ends[i], self.cutoff_ends[i + 1]
         __A									= d_embed // (div_val**i)

         self.out_projs.append(nn.Parameter(torch.FloatTensor(_lowerCamelCase,       _lowerCamelCase  )  )  )

         self.out_layers.append(nn.Linear(_lowerCamelCase,       r_idx - l_idx  )  )

     __A									= keep_order


   def 				_SCREAMING_SNAKE_CASE				(	self :     List[str],       _lowerCamelCase :     Optional[int],       _lowerCamelCase :     int,       _lowerCamelCase :     Optional[Any],       _lowerCamelCase :     List[str]  ):

     '''simple docstring'''
     if proj is None:
       __A									= nn.functional.linear(_lowerCamelCase,       _lowerCamelCase,       bias=_lowerCamelCase  )
     else:
       # if CUDA_MAJOR <= 9 and CUDA_MINOR <= 1:
       __A									= nn.functional.linear(_lowerCamelCase,       proj.t().contiguous()  )
       __A									= nn.functional.linear(_lowerCamelCase,       _lowerCamelCase,       bias=_lowerCamelCase  )
       # else:
       #     logit = torch.einsum('bd,de,ev->bv', (hidden, proj, weight.t()))
       #     if bias is not None:
       #         logit = logit + bias

     return logit


   def 				_SCREAMING_SNAKE_CASE				(	self :     Tuple,       _lowerCamelCase :     List[str],       _lowerCamelCase :     Tuple=None,       _lowerCamelCase :     Tuple=False  ):

     '''simple docstring'''

     if labels is not None:
       # Shift so that tokens < n predict n
       __A									= hidden[..., :-1, :].contiguous()
       __A									= labels[..., 1:].contiguous()
       __A									= hidden.view(-1,       hidden.size(-1  )  )
       __A									= labels.view(-1  )
       if hidden.size(0  ) != labels.size(0  ):
         raise RuntimeError('''Input and labels should have the same size in the batch dimension.'''  )
     else:
       __A									= hidden.view(-1,       hidden.size(-1  )  )

     if self.n_clusters == 0:
       __A									= self._compute_logit(_lowerCamelCase,       self.out_layers[0].weight,       self.out_layers[0].bias,       self.out_projs[0]  )
       if labels is not None:
         __A									= labels != -1_00
         __A									= torch.zeros_like(_lowerCamelCase,       dtype=hidden.dtype,       device=hidden.device  )
         __A									= (
             -nn.functional.log_softmax(_lowerCamelCase,       dim=-1  )[mask].gather(1,       labels[mask].unsqueeze(1  )  ).squeeze(1  )
         )
       else:
         __A									= nn.functional.log_softmax(_lowerCamelCase,       dim=-1  )
     else:
       # construct weights and biases
       __A			,    __A									= [], []
       for i in range(len(self.cutoffs  )  ):
         if self.div_val == 1:
           __A			,    __A									= self.cutoff_ends[i], self.cutoff_ends[i + 1]
           __A									= self.out_layers[0].weight[l_idx:r_idx]
           __A									= self.out_layers[0].bias[l_idx:r_idx]
         else:
           __A									= self.out_layers[i].weight
           __A									= self.out_layers[i].bias

         if i == 0:
           __A									= torch.cat([weight_i, self.cluster_weight],       dim=0  )
           __A									= torch.cat([bias_i, self.cluster_bias],       dim=0  )

         weights.append(_lowerCamelCase  )
         biases.append(_lowerCamelCase  )

       __A			,    __A			,    __A									= weights[0], biases[0], self.out_projs[0]

       __A									= self._compute_logit(_lowerCamelCase,       _lowerCamelCase,       _lowerCamelCase,       _lowerCamelCase  )
       __A									= nn.functional.log_softmax(_lowerCamelCase,       dim=1  )

       if labels is None:
         __A									= hidden.new_empty((head_logit.size(0  ), self.n_token)  )
       else:
         __A									= torch.zeros_like(_lowerCamelCase,       dtype=hidden.dtype,       device=hidden.device  )

       __A									= 0
       __A									= [0] + self.cutoffs
       for i in range(len(_lowerCamelCase  ) - 1  ):
         __A			,    __A									= cutoff_values[i], cutoff_values[i + 1]

         if labels is not None:
           __A									= (labels >= l_idx) & (labels < r_idx)
           __A									= mask_i.nonzero().squeeze()

           if indices_i.numel() == 0:
             continue

           __A									= labels.index_select(0,       _lowerCamelCase  ) - l_idx
           __A									= head_logprob.index_select(0,       _lowerCamelCase  )
           __A									= hidden.index_select(0,       _lowerCamelCase  )
         else:
           __A									= hidden

         if i == 0:
           if labels is not None:
             __A									= head_logprob_i.gather(1,       target_i[:, None]  ).squeeze(1  )
           else:
             __A									= head_logprob[:, : self.cutoffs[0]]
         else:
           __A			,    __A			,    __A									= weights[i], biases[i], self.out_projs[i]

           __A									= self._compute_logit(_lowerCamelCase,       _lowerCamelCase,       _lowerCamelCase,       _lowerCamelCase  )
           __A									= nn.functional.log_softmax(_lowerCamelCase,       dim=1  )
           __A									= self.cutoffs[0] + i - 1  # No probability for the head cluster
           if labels is not None:
             __A									= head_logprob_i[:, cluster_prob_idx] + tail_logprob_i.gather(
                 1,       target_i[:, None]  ).squeeze(1  )
           else:
             __A									= head_logprob[:, cluster_prob_idx, None] + tail_logprob_i
             __A									= logprob_i

         if labels is not None:
           if (hasattr(self,       '''keep_order'''  ) and self.keep_order) or keep_order:
             out.index_copy_(0,       _lowerCamelCase,       -logprob_i  )
           else:
             out[offset : offset + logprob_i.size(0  )].copy_(-logprob_i  )
           offset += logprob_i.size(0  )

     return out






   def 				_SCREAMING_SNAKE_CASE				(	self :     Any,       _lowerCamelCase :     Optional[int]  ):

     '''simple docstring'''
     if self.n_clusters == 0:
       __A									= self._compute_logit(_lowerCamelCase,       self.out_layers[0].weight,       self.out_layers[0].bias,       self.out_projs[0]  )
       return nn.functional.log_softmax(_lowerCamelCase,       dim=-1  )
     else:
       # construct weights and biases
       __A			,    __A									= [], []
       for i in range(len(self.cutoffs  )  ):
         if self.div_val == 1:
           __A			,    __A									= self.cutoff_ends[i], self.cutoff_ends[i + 1]
           __A									= self.out_layers[0].weight[l_idx:r_idx]
           __A									= self.out_layers[0].bias[l_idx:r_idx]
         else:
           __A									= self.out_layers[i].weight
           __A									= self.out_layers[i].bias

         if i == 0:
           __A									= torch.cat([weight_i, self.cluster_weight],       dim=0  )
           __A									= torch.cat([bias_i, self.cluster_bias],       dim=0  )

         weights.append(_lowerCamelCase  )
         biases.append(_lowerCamelCase  )

       __A			,    __A			,    __A									= weights[0], biases[0], self.out_projs[0]
       __A									= self._compute_logit(_lowerCamelCase,       _lowerCamelCase,       _lowerCamelCase,       _lowerCamelCase  )

       __A									= hidden.new_empty((head_logit.size(0  ), self.n_token)  )
       __A									= nn.functional.log_softmax(_lowerCamelCase,       dim=1  )

       __A									= [0] + self.cutoffs
       for i in range(len(_lowerCamelCase  ) - 1  ):
         __A			,    __A									= cutoff_values[i], cutoff_values[i + 1]

         if i == 0:
           __A									= head_logprob[:, : self.cutoffs[0]]
         else:
           __A			,    __A			,    __A									= weights[i], biases[i], self.out_projs[i]

           __A									= self._compute_logit(_lowerCamelCase,       _lowerCamelCase,       _lowerCamelCase,       _lowerCamelCase  )
           __A									= nn.functional.log_softmax(_lowerCamelCase,       dim=1  )

           __A									= head_logprob[:, -i] + tail_logprob_i
           __A									= logprob_i

       return out












"""simple docstring"""


import itertools
import random
import unittest

import numpy as np

from transformers import ASTFeatureExtractor
from transformers.testing_utils import require_torch, require_torchaudio
from transformers.utils.import_utils import is_torch_available

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


lowercase_				     =     random.Random()

if is_torch_available():
       import torch



def 					lowerCAmelCase				(    __UpperCamelCase		,       __UpperCamelCase=1.0		,       __UpperCamelCase=None		,       __UpperCamelCase=None ):




  """simple docstring"""







  if rng is None:
    __A									= global_rng

  __A									= []
  for batch_idx in range(shape[0] ):
    values.append([] )
    for _ in range(shape[1] ):
      values[-1].append(rng.random() * scale )

  return values


class 					snake_case		(							unittest.TestCase						):


   '''simple docstring'''

   def __init__(	self :     Any,       _lowerCamelCase :     List[str],       _lowerCamelCase :     Any=7,       _lowerCamelCase :     Optional[int]=4_00,       _lowerCamelCase :     Optional[int]=20_00,       _lowerCamelCase :     Dict=1,       _lowerCamelCase :     Optional[Any]=0.0,       _lowerCamelCase :     int=1_60_00,       _lowerCamelCase :     Optional[int]=True,       _lowerCamelCase :     Dict=True,       ):

     '''simple docstring'''
     __A									= parent
     __A									= batch_size
     __A									= min_seq_length
     __A									= max_seq_length
     __A									= (self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
     __A									= feature_size
     __A									= padding_value
     __A									= sampling_rate
     __A									= return_attention_mask
     __A									= do_normalize


   def 				_SCREAMING_SNAKE_CASE				(	self :     Union[str, Any]  ):

     '''simple docstring'''
     return {
         "feature_size": self.feature_size,
         "padding_value": self.padding_value,
         "sampling_rate": self.sampling_rate,
         "return_attention_mask": self.return_attention_mask,
         "do_normalize": self.do_normalize,
     }






   def 				_SCREAMING_SNAKE_CASE				(	self :     Any,       _lowerCamelCase :     Optional[Any]=False,       _lowerCamelCase :     int=False  ):

     '''simple docstring'''
     def _flatten(_lowerCamelCase :     List[str]  ):
       return list(itertools.chain(*_lowerCamelCase  )  )

     if equal_length:
       __A									= floats_list((self.batch_size, self.max_seq_length)  )
     else:
       # make sure that inputs increase in size
       __A									= [
           _flatten(floats_list((x, self.feature_size)  )  )
           for x in range(self.min_seq_length,       self.max_seq_length,       self.seq_length_diff  )
       ]

     if numpify:
       __A									= [np.asarray(_lowerCamelCase  ) for x in speech_inputs]

     return speech_inputs





@require_torch
@require_torchaudio
class 					snake_case		(							_lowerCAmelCase   ,	unittest.TestCase						):


   '''simple docstring'''

   A_						:	int						 = ASTFeatureExtractor


   def 				_SCREAMING_SNAKE_CASE				(	self :     Tuple  ):

     '''simple docstring'''
     __A									= ASTFeatureExtractionTester(self  )


   def 				_SCREAMING_SNAKE_CASE				(	self :     Dict  ):

     '''simple docstring'''
     # Tests that all call wrap to encode_plus and batch_encode_plus
     __A									= self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()  )
     # create three inputs of length 800, 1000, and 1200
     __A									= [floats_list((1, x)  )[0] for x in range(8_00,       14_00,       2_00  )]
     __A									= [np.asarray(_lowerCamelCase  ) for speech_input in speech_inputs]

     # Test not batched input
     __A									= feat_extract(speech_inputs[0],       return_tensors='''np'''  ).input_values
     __A									= feat_extract(np_speech_inputs[0],       return_tensors='''np'''  ).input_values
     self.assertTrue(np.allclose(_lowerCamelCase,       _lowerCamelCase,       atol=1e-3  )  )

     # Test batched
     __A									= feat_extract(_lowerCamelCase,       padding=_lowerCamelCase,       return_tensors='''np'''  ).input_values
     __A									= feat_extract(_lowerCamelCase,       padding=_lowerCamelCase,       return_tensors='''np'''  ).input_values
     for enc_seq_a, enc_seq_a in zip(_lowerCamelCase,       _lowerCamelCase  ):
       self.assertTrue(np.allclose(_lowerCamelCase,       _lowerCamelCase,       atol=1e-3  )  )

     # Test 2-D numpy arrays are batched.
     __A									= [floats_list((1, x)  )[0] for x in (8_00, 8_00, 8_00)]
     __A									= np.asarray(_lowerCamelCase  )
     __A									= feat_extract(_lowerCamelCase,       return_tensors='''np'''  ).input_values
     __A									= feat_extract(_lowerCamelCase,       return_tensors='''np'''  ).input_values
     for enc_seq_a, enc_seq_a in zip(_lowerCamelCase,       _lowerCamelCase  ):
       self.assertTrue(np.allclose(_lowerCamelCase,       _lowerCamelCase,       atol=1e-3  )  )


   @require_torch
   def 				_SCREAMING_SNAKE_CASE				(	self :     Union[str, Any]  ):

     '''simple docstring'''
     import torch

     __A									= self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()  )
     __A									= np.random.rand(1_00  ).astype(np.floataa  )
     __A									= np_speech_inputs.tolist()

     for inputs in [py_speech_inputs, np_speech_inputs]:
       __A									= feature_extractor.pad([{'''input_values''': inputs}],       return_tensors='''np'''  )
       self.assertTrue(np_processed.input_values.dtype == np.floataa  )
       __A									= feature_extractor.pad([{'''input_values''': inputs}],       return_tensors='''pt'''  )
       self.assertTrue(pt_processed.input_values.dtype == torch.floataa  )


   def 				_SCREAMING_SNAKE_CASE				(	self :     Optional[int],       _lowerCamelCase :     Union[str, Any]  ):

     '''simple docstring'''
     from datasets import load_dataset

     __A									= load_dataset('''hf-internal-testing/librispeech_asr_dummy''',       '''clean''',       split='''validation'''  )
     # automatic decoding with librispeech
     __A									= ds.sort('''id'''  ).select(range(_lowerCamelCase  )  )[:num_samples]['''audio''']

     return [x["array"] for x in speech_samples]






   @require_torch
   def 				_SCREAMING_SNAKE_CASE				(	self :     Optional[Any]  ):

     '''simple docstring'''
     # fmt: off
     __A									= torch.tensor(
         [-0.98_94, -1.27_76, -0.90_66, -1.27_76, -0.93_49, -1.26_09, -1.03_86, -1.27_76,
          -1.15_61, -1.27_76, -1.20_52, -1.27_23, -1.21_90, -1.21_32, -1.27_76, -1.11_33,
          -1.19_53, -1.13_43, -1.15_84, -1.22_03, -1.17_70, -1.24_74, -1.23_81, -1.19_36,
          -0.92_70, -0.83_17, -0.80_49, -0.77_06, -0.75_65, -0.78_69]  )
     # fmt: on

     __A									= self._load_datasamples(1  )
     __A									= ASTFeatureExtractor()
     __A									= feature_extractor(_lowerCamelCase,       return_tensors='''pt'''  ).input_values
     self.assertEquals(input_values.shape,       (1, 10_24, 1_28)  )
     self.assertTrue(torch.allclose(input_values[0, 0, :30],       _lowerCamelCase,       atol=1e-4  )  )




