


'''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 TensorType, logging


a__  : Optional[int]				=	logging.get_logger(__name__)
class  lowercase_					(   a__       ):
    __UpperCAmelCase             =     ['input_values', 'attention_mask']


    def __init__(					self  , a=1  , a=1_60_00  , a=1_28  , a=10_24  , a=0.0  , a=True  , a=-4.267_7393  , a=4.568_9974  , a=False  , **a  , ):
          super().__init__(feature_size=a  , sampling_rate=a  , padding_value=a  , **a   )
          UpperCamelCase__  =    num_mel_bins
          UpperCamelCase__  =    max_length
          UpperCamelCase__  =    do_normalize
          UpperCamelCase__  =    mean
          UpperCamelCase__  =    std
          UpperCamelCase__  =    return_attention_mask


    def        __a							(					self  , a  , a  , ):
          UpperCamelCase__  =    torch.from_numpy(a   ).unsqueeze(0   )
          UpperCamelCase__  =    ta_kaldi.fbank(
              a  , htk_compat=a  , sample_frequency=self.sampling_rate  , use_energy=a  , window_type="hanning"  , num_mel_bins=self.num_mel_bins  , dither=0.0  , frame_shift=10  , )

          UpperCamelCase__  =    fbank.shape[0]
          UpperCamelCase__  =    max_length - n_frames

          # pad or truncate, depending on difference
          if difference > 0:
                UpperCamelCase__  =    torch.nn.ZeroPadad((0, 0, 0, difference)   )
                UpperCamelCase__  =    pad_module(a   )
          elif difference < 0:
                UpperCamelCase__  =    fbank[0:max_length, :]

          UpperCamelCase__  =    fbank.numpy()

          return fbank


    def        __a							(					self  , a   ):
          return (input_values - (self.mean)) / (self.std * 2)


    def __call__(					self  , a  , a = None  , a = None  , **a  , ):

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

          UpperCamelCase__  =    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}'''   )
          UpperCamelCase__  =    is_batched_numpy or (
              isinstance(a  , (list, tuple)   ) and (isinstance(raw_speech[0]  , (np.ndarray, tuple, list)   ))
          )

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

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

          # extract fbank features and pad/truncate to max_length
          UpperCamelCase__  =    [self._extract_fbank_features(a  , max_length=self.max_length   ) for waveform in raw_speech]

          # convert into BatchFeature
          UpperCamelCase__  =    BatchFeature({"input_values": features}   )

          # make sure list is in array format
          UpperCamelCase__  =    padded_inputs.get("input_values"   )
          if isinstance(input_values[0]  , a   ):
                UpperCamelCase__  =    [np.asarray(a  , dtype=np.floataa   ) for feature in input_values]

          # normalization
          if self.do_normalize:
                UpperCamelCase__  =    [self.normalize(a   ) for feature in input_values]

          if return_tensors is not None:
                UpperCamelCase__  =    padded_inputs.convert_to_tensors(a   )

          return padded_inputs









'''simple docstring'''







import json
import os
import shutil
import tempfile
import unittest

from transformers import is_speech_available, is_torch_available
from transformers.models.speechta import SpeechTaTokenizer
from transformers.testing_utils import get_tests_dir, require_torch
from transformers.utils import FEATURE_EXTRACTOR_NAME


if is_speech_available() and is_torch_available():
  from transformers import SpeechTaFeatureExtractor, SpeechTaProcessor

  from .test_feature_extraction_speechta import floats_list


a__  : int				=	get_tests_dir('fixtures/test_sentencepiece_bpe_char.model')
@require_torch
class  lowercase_					(   unittest.TestCase       ):


    def        __a							(					self   ):
          UpperCamelCase__  =    tempfile.mkdtemp()

          UpperCamelCase__  =    SpeechTaTokenizer(a   )
          tokenizer.save_pretrained(self.tmpdirname   )

          UpperCamelCase__  =    {
              "feature_size": 1,
              "padding_value": 0.0,
              "sampling_rate": 1_60_00,
              "do_normalize": False,
              "num_mel_bins": 80,
              "hop_length": 16,
              "win_length": 64,
              "win_function": "hann_window",
              "fmin": 80,
              "fmax": 76_00,
              "mel_floor": 1e-10,
              "reduction_factor": 2,
              "return_attention_mask": True,
          }

          UpperCamelCase__  =    os.path.join(self.tmpdirname  , a   )
          with open(self.feature_extraction_file  , "w"  , encoding="utf-8"   ) as fp:
                fp.write(json.dumps(a   ) + "\n"   )


    def        __a							(					self  , **a   ):
          return SpeechTaTokenizer.from_pretrained(self.tmpdirname  , **a   )


    def        __a							(					self  , **a   ):
          return SpeechTaFeatureExtractor.from_pretrained(self.tmpdirname  , **a   )


    def        __a							(					self   ):
          shutil.rmtree(self.tmpdirname   )


    def        __a							(					self   ):
          UpperCamelCase__  =    self.get_tokenizer()
          UpperCamelCase__  =    self.get_feature_extractor()

          UpperCamelCase__  =    SpeechTaProcessor(tokenizer=a  , feature_extractor=a   )

          processor.save_pretrained(self.tmpdirname   )
          UpperCamelCase__  =    SpeechTaProcessor.from_pretrained(self.tmpdirname   )

          self.assertEqual(processor.tokenizer.get_vocab()  , tokenizer.get_vocab()   )
          self.assertIsInstance(processor.tokenizer  , a   )

          self.assertEqual(processor.feature_extractor.to_json_string()  , feature_extractor.to_json_string()   )
          self.assertIsInstance(processor.feature_extractor  , a   )


    def        __a							(					self   ):
          UpperCamelCase__  =    SpeechTaProcessor(tokenizer=self.get_tokenizer()  , feature_extractor=self.get_feature_extractor()   )
          processor.save_pretrained(self.tmpdirname   )

          UpperCamelCase__  =    self.get_tokenizer(bos_token="(BOS)"  , eos_token="(EOS)"   )
          UpperCamelCase__  =    self.get_feature_extractor(do_normalize=a  , padding_value=1.0   )

          UpperCamelCase__  =    SpeechTaProcessor.from_pretrained(
              self.tmpdirname  , bos_token="(BOS)"  , eos_token="(EOS)"  , do_normalize=a  , padding_value=1.0   )

          self.assertEqual(processor.tokenizer.get_vocab()  , tokenizer_add_kwargs.get_vocab()   )
          self.assertIsInstance(processor.tokenizer  , a   )

          self.assertEqual(processor.feature_extractor.to_json_string()  , feature_extractor_add_kwargs.to_json_string()   )
          self.assertIsInstance(processor.feature_extractor  , a   )


    def        __a							(					self   ):
          UpperCamelCase__  =    self.get_feature_extractor()
          UpperCamelCase__  =    self.get_tokenizer()

          UpperCamelCase__  =    SpeechTaProcessor(tokenizer=a  , feature_extractor=a   )

          UpperCamelCase__  =    floats_list((3, 10_00)   )

          UpperCamelCase__  =    feature_extractor(audio=a  , return_tensors="np"   )
          UpperCamelCase__  =    processor(audio=a  , return_tensors="np"   )

          for key in input_feat_extract.keys():
                self.assertAlmostEqual(input_feat_extract[key].sum()  , input_processor[key].sum()  , delta=1e-2   )


    def        __a							(					self   ):
          UpperCamelCase__  =    self.get_feature_extractor()
          UpperCamelCase__  =    self.get_tokenizer()

          UpperCamelCase__  =    SpeechTaProcessor(tokenizer=a  , feature_extractor=a   )

          UpperCamelCase__  =    floats_list((3, 10_00)   )

          UpperCamelCase__  =    feature_extractor(audio_target=a  , return_tensors="np"   )
          UpperCamelCase__  =    processor(audio_target=a  , return_tensors="np"   )

          for key in input_feat_extract.keys():
                self.assertAlmostEqual(input_feat_extract[key].sum()  , input_processor[key].sum()  , delta=1e-2   )


    def        __a							(					self   ):
          UpperCamelCase__  =    self.get_feature_extractor()
          UpperCamelCase__  =    self.get_tokenizer()

          UpperCamelCase__  =    SpeechTaProcessor(tokenizer=a  , feature_extractor=a   )

          UpperCamelCase__  =    "This is a test string"

          UpperCamelCase__  =    processor(text=a   )
          UpperCamelCase__  =    tokenizer(a   )

          for key in encoded_tok.keys():
                self.assertListEqual(encoded_tok[key]  , encoded_processor[key]   )


    def        __a							(					self   ):
          UpperCamelCase__  =    self.get_feature_extractor()
          UpperCamelCase__  =    self.get_tokenizer()

          UpperCamelCase__  =    SpeechTaProcessor(tokenizer=a  , feature_extractor=a   )

          UpperCamelCase__  =    "This is a test string"

          UpperCamelCase__  =    processor(text_target=a   )
          UpperCamelCase__  =    tokenizer(a   )

          for key in encoded_tok.keys():
                self.assertListEqual(encoded_tok[key]  , encoded_processor[key]   )


    def        __a							(					self   ):
          UpperCamelCase__  =    self.get_feature_extractor()
          UpperCamelCase__  =    self.get_tokenizer()

          UpperCamelCase__  =    SpeechTaProcessor(tokenizer=a  , feature_extractor=a   )

          UpperCamelCase__  =    [[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

          UpperCamelCase__  =    processor.batch_decode(a   )
          UpperCamelCase__  =    tokenizer.batch_decode(a   )

          self.assertListEqual(a  , a   )


    def        __a							(					self   ):
          UpperCamelCase__  =    self.get_feature_extractor()
          UpperCamelCase__  =    self.get_tokenizer()

          UpperCamelCase__  =    SpeechTaProcessor(tokenizer=a  , feature_extractor=a   )

          self.assertListEqual(
              processor.model_input_names  , feature_extractor.model_input_names  , msg="`processor` and `feature_extractor` model input names do not match"  , )





