




import json
import os
import shutil
import tempfile
import unittest
from multiprocessing import get_context
from pathlib import Path

import datasets
import numpy as np
from datasets import load_dataset
from parameterized import parameterized

from transformers import AutoProcessor
from transformers.models.wavaveca import WavaVecaCTCTokenizer, WavaVecaFeatureExtractor
from transformers.models.wavaveca.tokenization_wavaveca import VOCAB_FILES_NAMES
from transformers.testing_utils import require_pyctcdecode, require_torch, require_torchaudio, slow
from transformers.utils import FEATURE_EXTRACTOR_NAME, is_pyctcdecode_available, is_torch_available

from ..wavaveca.test_feature_extraction_wavaveca import floats_list


if is_pyctcdecode_available():
  from huggingface_hub import snapshot_download
  from pyctcdecode import BeamSearchDecoderCTC

  from transformers.models.wavaveca_with_lm import WavaVecaProcessorWithLM
  from transformers.models.wavaveca_with_lm.processing_wavaveca_with_lm import WavaVecaDecoderWithLMOutput

if is_torch_available():
  from transformers import WavaVecaForCTC







@require_pyctcdecode
class 						lowercase							(   unittest.TestCase		):

     def 					a__	(  self				)						->							str:
           _A							:     Any			=						"""| <pad> <unk> <s> </s> a b c d e f g h i j k""".split()
           _A							:     Tuple			=						dict(zip(_a		,     range(len(_a				)				)				)				)

           _A							:     List[str]			=						{
               """unk_token""": """<unk>""",
               """bos_token""": """<s>""",
               """eos_token""": """</s>""",
           }
           _A							:     Optional[Any]			=						{
               """feature_size""": 1,
               """padding_value""": 0.0,
               """sampling_rate""": 1_6000,
               """return_attention_mask""": False,
               """do_normalize""": True,
           }

           _A							:     Union[str, Any]			=						tempfile.mkdtemp()
           _A							:     Union[str, Any]			=						os.path.join(self.tmpdirname		,     VOCAB_FILES_NAMES["""vocab_file"""]				)
           _A							:     Optional[Any]			=						os.path.join(self.tmpdirname		,     _a				)
           with open(self.vocab_file		,     """w"""		,     encoding="""utf-8"""				) as fp:
                 fp.write(json.dumps(_a				) + """\n"""				)

           with open(self.feature_extraction_file		,     """w"""		,     encoding="""utf-8"""				) as fp:
                 fp.write(json.dumps(_a				) + """\n"""				)

           # load decoder from hub
           _A							:     str			=						"""hf-internal-testing/ngram-beam-search-decoder"""

     def 					a__	(  self		,     **_a				)						->							Union[str, Any]:
           _A							:     str			=						self.add_kwargs_tokens_map.copy()
           kwargs.update(_a				)
           return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname		,     **_a				)

     def 					a__	(  self		,     **_a				)						->							Dict:
           return WavaVecaFeatureExtractor.from_pretrained(self.tmpdirname		,     **_a				)

     def 					a__	(  self		,     **_a				)						->							str:
           return BeamSearchDecoderCTC.load_from_hf_hub(self.decoder_name		,     **_a				)

     def 					a__	(  self				)						->							Any:
           shutil.rmtree(self.tmpdirname				)

     def 					a__	(  self				)						->							str:
           _A							:     List[Any]			=						self.get_tokenizer()
           _A							:     Optional[Any]			=						self.get_feature_extractor()
           _A							:     List[str]			=						self.get_decoder()

           _A							:     List[str]			=						WavaVecaProcessorWithLM(tokenizer=_a		,     feature_extractor=_a		,     decoder=_a				)

           processor.save_pretrained(self.tmpdirname				)
           _A							:     List[str]			=						WavaVecaProcessorWithLM.from_pretrained(self.tmpdirname				)

           # tokenizer
           self.assertEqual(processor.tokenizer.get_vocab()		,     tokenizer.get_vocab()				)
           self.assertIsInstance(processor.tokenizer		,     _a				)

           # feature extractor
           self.assertEqual(processor.feature_extractor.to_json_string()		,     feature_extractor.to_json_string()				)
           self.assertIsInstance(processor.feature_extractor		,     _a				)

           # decoder
           self.assertEqual(processor.decoder._alphabet.labels		,     decoder._alphabet.labels				)
           self.assertEqual(
               processor.decoder.model_container[decoder._model_key]._unigram_set		,     decoder.model_container[decoder._model_key]._unigram_set		,     )
           self.assertIsInstance(processor.decoder		,     _a				)

     def 					a__	(  self				)						->							Union[str, Any]:
           _A							:     List[Any]			=						WavaVecaProcessorWithLM(
               tokenizer=self.get_tokenizer()		,     feature_extractor=self.get_feature_extractor()		,     decoder=self.get_decoder()				)
           processor.save_pretrained(self.tmpdirname				)

           # make sure that error is thrown when decoder alphabet doesn't match
           _A							:     int			=						WavaVecaProcessorWithLM.from_pretrained(
               self.tmpdirname		,     alpha=5.0		,     beta=3.0		,     score_boundary=-7.0		,     unk_score_offset=3				)

           # decoder
           self.assertEqual(processor.language_model.alpha		,     5.0				)
           self.assertEqual(processor.language_model.beta		,     3.0				)
           self.assertEqual(processor.language_model.score_boundary		,     -7.0				)
           self.assertEqual(processor.language_model.unk_score_offset		,     3				)

     def 					a__	(  self				)						->							Optional[Any]:
           _A							:     Dict			=						self.get_tokenizer()
           # add token to trigger raise
           tokenizer.add_tokens(["""xx"""]				)
           with self.assertRaisesRegex(_a		,     """include"""				):
                 WavaVecaProcessorWithLM(
                     tokenizer=_a		,     feature_extractor=self.get_feature_extractor()		,     decoder=self.get_decoder()				)

     def 					a__	(  self				)						->							List[str]:
           _A							:     str			=						self.get_feature_extractor()
           _A							:     Tuple			=						self.get_tokenizer()
           _A							:     List[Any]			=						self.get_decoder()

           _A							:     Any			=						WavaVecaProcessorWithLM(tokenizer=_a		,     feature_extractor=_a		,     decoder=_a				)

           _A							:     Tuple			=						floats_list((3, 1000)				)

           _A							:     Union[str, Any]			=						feature_extractor(_a		,     return_tensors="""np"""				)
           _A							:     Tuple			=						processor(_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				)						->							str:
           _A							:     Any			=						self.get_feature_extractor()
           _A							:     List[str]			=						self.get_tokenizer()
           _A							:     List[Any]			=						self.get_decoder()

           _A							:     List[str]			=						WavaVecaProcessorWithLM(tokenizer=_a		,     feature_extractor=_a		,     decoder=_a				)

           _A							:     int			=						"""This is a test string"""

           _A							:     List[str]			=						processor(text=_a				)

           _A							:     Union[str, Any]			=						tokenizer(_a				)

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

     def 					a__	(  self		,     _a=(2, 10, 16)		,     _a=77				)						->							Optional[int]:
           np.random.seed(_a				)
           return np.random.rand(*_a				)

     def 					a__	(  self				)						->							List[str]:
           _A							:     Dict			=						self.get_feature_extractor()
           _A							:     List[Any]			=						self.get_tokenizer()
           _A							:     Optional[int]			=						self.get_decoder()

           _A							:     Optional[Any]			=						WavaVecaProcessorWithLM(tokenizer=_a		,     feature_extractor=_a		,     decoder=_a				)

           _A							:     List[Any]			=						self._get_dummy_logits(shape=(10, 16)		,     seed=13				)

           _A							:     Dict			=						processor.decode(_a				)

           _A							:     Optional[Any]			=						decoder.decode_beams(_a				)[0]

           self.assertEqual(decoded_decoder[0]		,     decoded_processor.text				)
           self.assertEqual("""</s> <s> </s>"""		,     decoded_processor.text				)
           self.assertEqual(decoded_decoder[-2]		,     decoded_processor.logit_score				)
           self.assertEqual(decoded_decoder[-1]		,     decoded_processor.lm_score				)

     @parameterized.expand([[None], ["""fork"""], ["""spawn"""]]				)
     def 					a__	(  self		,     _a				)						->							int:
           _A							:     int			=						self.get_feature_extractor()
           _A							:     Any			=						self.get_tokenizer()
           _A							:     int			=						self.get_decoder()

           _A							:     Tuple			=						WavaVecaProcessorWithLM(tokenizer=_a		,     feature_extractor=_a		,     decoder=_a				)

           _A							:     Union[str, Any]			=						self._get_dummy_logits()

           # note: pool should be instantiated *after* Wav2Vec2ProcessorWithLM.
           #       otherwise, the LM won't be available to the pool's sub-processes.
           # manual logic used to allow parameterized test for both pool=None and pool=Pool(...)
           if pool_context is None:
                 _A							:     Any			=						processor.batch_decode(_a				)
           else:
                 with get_context(_a				).Pool() as pool:
                       _A							:     Optional[int]			=						processor.batch_decode(_a		,     _a				)

           _A							:     Optional[int]			=						list(_a				)

           with get_context("""fork"""				).Pool() as p:
                 _A							:     Tuple			=						decoder.decode_beams_batch(_a		,     _a				)

           _A			,   _A			,   _A							:     str			=						[], [], []
           for beams in decoded_beams:
                 texts_decoder.append(beams[0][0]				)
                 logit_scores_decoder.append(beams[0][-2]				)
                 lm_scores_decoder.append(beams[0][-1]				)

           self.assertListEqual(_a		,     decoded_processor.text				)
           self.assertListEqual(["""<s> <s> </s>""", """<s> <s> <s>"""]		,     decoded_processor.text				)
           self.assertListEqual(_a		,     decoded_processor.logit_score				)
           self.assertListEqual(_a		,     decoded_processor.lm_score				)

     def 					a__	(  self				)						->							Optional[Any]:
           _A							:     Any			=						self.get_feature_extractor()
           _A							:     str			=						self.get_tokenizer()
           _A							:     int			=						self.get_decoder()

           _A							:     Dict			=						WavaVecaProcessorWithLM(tokenizer=_a		,     feature_extractor=_a		,     decoder=_a				)

           _A							:     List[Any]			=						self._get_dummy_logits()

           _A							:     Union[str, Any]			=						15
           _A							:     str			=						-20.0
           _A							:     Optional[int]			=						-4.0

           _A							:     Union[str, Any]			=						processor.batch_decode(
               _a		,     beam_width=_a		,     beam_prune_logp=_a		,     token_min_logp=_a		,     )
           _A							:     str			=						decoded_processor_out.text

           _A							:     Dict			=						list(_a				)

           with get_context("""fork"""				).Pool() as pool:
                 _A							:     Union[str, Any]			=						decoder.decode_beams_batch(
                     _a		,     _a		,     beam_width=_a		,     beam_prune_logp=_a		,     token_min_logp=_a		,     )

           _A							:     Optional[int]			=						[d[0][0] for d in decoded_decoder_out]
           _A							:     int			=						[d[0][2] for d in decoded_decoder_out]
           _A							:     Optional[int]			=						[d[0][3] for d in decoded_decoder_out]

           self.assertListEqual(_a		,     _a				)
           self.assertListEqual(["""</s> <s> <s>""", """<s> <s> <s>"""]		,     _a				)

           self.assertTrue(np.array_equal(_a		,     decoded_processor_out.logit_score				)				)
           self.assertTrue(np.allclose([-20.054, -18.447]		,     _a		,     atol=1e-3				)				)

           self.assertTrue(np.array_equal(_a		,     decoded_processor_out.lm_score				)				)
           self.assertTrue(np.allclose([-15.554, -13.9474]		,     _a		,     atol=1e-3				)				)

     def 					a__	(  self				)						->							str:
           _A							:     Any			=						self.get_feature_extractor()
           _A							:     Dict			=						self.get_tokenizer()
           _A							:     Any			=						self.get_decoder()

           _A							:     Any			=						WavaVecaProcessorWithLM(tokenizer=_a		,     feature_extractor=_a		,     decoder=_a				)

           _A							:     Dict			=						self._get_dummy_logits()

           _A							:     Any			=						2.0
           _A							:     int			=						5.0
           _A							:     Union[str, Any]			=						-20.0
           _A							:     str			=						True

           _A							:     Optional[Any]			=						processor.batch_decode(
               _a		,     alpha=_a		,     beta=_a		,     unk_score_offset=_a		,     lm_score_boundary=_a		,     )
           _A							:     Union[str, Any]			=						decoded_processor_out.text

           _A							:     Tuple			=						list(_a				)
           decoder.reset_params(
               alpha=_a		,     beta=_a		,     unk_score_offset=_a		,     lm_score_boundary=_a		,     )

           with get_context("""fork"""				).Pool() as pool:
                 _A							:     Optional[Any]			=						decoder.decode_beams_batch(
                     _a		,     _a		,     )

           _A							:     Dict			=						[d[0][0] for d in decoded_decoder_out]

           self.assertListEqual(_a		,     _a				)
           self.assertListEqual(["""<s> </s> <s> </s> </s>""", """</s> </s> <s> </s> </s>"""]		,     _a				)
           _A							:     List[Any]			=						processor.decoder.model_container[processor.decoder._model_key]
           self.assertEqual(lm_model.alpha		,     2.0				)
           self.assertEqual(lm_model.beta		,     5.0				)
           self.assertEqual(lm_model.unk_score_offset		,     -20.0				)
           self.assertEqual(lm_model.score_boundary		,     _a				)

     def 					a__	(  self				)						->							Any:
           _A							:     Tuple			=						WavaVecaProcessorWithLM.from_pretrained("""hf-internal-testing/processor_with_lm"""				)

           _A							:     Optional[int]			=						processor.decoder.model_container[processor.decoder._model_key]
           _A							:     Optional[int]			=						Path(language_model._kenlm_model.path.decode("""utf-8"""				)				).parent.parent.absolute()

           _A							:     Any			=						os.listdir(_a				)
           _A							:     int			=						["""alphabet.json""", """language_model"""]

           downloaded_decoder_files.sort()
           expected_decoder_files.sort()

           # test that only decoder relevant files from
           # https://huggingface.co/hf-internal-testing/processor_with_lm/tree/main
           # are downloaded and none of the rest (e.g. README.md, ...)
           self.assertListEqual(_a		,     _a				)

     def 					a__	(  self				)						->							Dict:
           _A							:     int			=						snapshot_download("""hf-internal-testing/processor_with_lm"""				)

           _A							:     Optional[Any]			=						WavaVecaProcessorWithLM.from_pretrained(_a				)

           _A							:     Optional[int]			=						processor.decoder.model_container[processor.decoder._model_key]
           _A							:     str			=						Path(language_model._kenlm_model.path.decode("""utf-8"""				)				).parent.parent.absolute()

           _A							:     Optional[int]			=						os.listdir(_a				)
           _A							:     List[str]			=						os.listdir(_a				)

           local_decoder_files.sort()
           expected_decoder_files.sort()

           # test that both decoder form hub and local files in cache are the same
           self.assertListEqual(_a		,     _a				)

     def 					a__	(  self				)						->							Optional[Any]:
           _A							:     List[str]			=						WavaVecaProcessorWithLM.from_pretrained("""hf-internal-testing/processor_with_lm"""				)
           _A							:     Union[str, Any]			=						AutoProcessor.from_pretrained("""hf-internal-testing/processor_with_lm"""				)

           _A							:     List[str]			=						floats_list((3, 1000)				)

           _A							:     Union[str, Any]			=						processor_wavaveca(_a		,     return_tensors="""np"""				)
           _A							:     Any			=						processor_auto(_a		,     return_tensors="""np"""				)

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

           _A							:     Tuple			=						self._get_dummy_logits()

           _A							:     List[str]			=						processor_wavaveca.batch_decode(_a				)
           _A							:     Any			=						processor_auto.batch_decode(_a				)

           self.assertListEqual(decoded_wavaveca.text		,     decoded_auto.text				)

     def 					a__	(  self				)						->							Dict:
           _A							:     Dict			=						self.get_feature_extractor()
           _A							:     Optional[int]			=						self.get_tokenizer()
           _A							:     int			=						self.get_decoder()

           _A							:     List[str]			=						WavaVecaProcessorWithLM(tokenizer=_a		,     feature_extractor=_a		,     decoder=_a				)

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

     @staticmethod
     def 					a__	(  _a		,     _a				)						->							int:
           _A							:     List[Any]			=						[d[key] for d in offsets]
           return retrieved_list

     def 					a__	(  self				)						->							int:
           _A							:     Any			=						WavaVecaProcessorWithLM.from_pretrained("""hf-internal-testing/processor_with_lm"""				)
           _A							:     Optional[int]			=						self._get_dummy_logits()[0]

           _A							:     str			=						processor.decode(_a		,     output_word_offsets=_a				)
           # check Wav2Vec2CTCTokenizerOutput keys for word
           self.assertEqual(len(outputs.keys()				)		,     4				)
           self.assertTrue("""text""" in outputs				)
           self.assertTrue("""word_offsets""" in outputs				)
           self.assertTrue(isinstance(_a		,     _a				)				)

           self.assertEqual(""" """.join(self.get_from_offsets(outputs["""word_offsets"""]		,     """word"""				)				)		,     outputs.text				)
           self.assertListEqual(self.get_from_offsets(outputs["""word_offsets"""]		,     """word"""				)		,     ["""<s>""", """<s>""", """</s>"""]				)
           self.assertListEqual(self.get_from_offsets(outputs["""word_offsets"""]		,     """start_offset"""				)		,     [0, 2, 4]				)
           self.assertListEqual(self.get_from_offsets(outputs["""word_offsets"""]		,     """end_offset"""				)		,     [1, 3, 5]				)

     def 					a__	(  self				)						->							int:
           _A							:     Dict			=						WavaVecaProcessorWithLM.from_pretrained("""hf-internal-testing/processor_with_lm"""				)
           _A							:     Tuple			=						self._get_dummy_logits()

           _A							:     int			=						processor.batch_decode(_a		,     output_word_offsets=_a				)

           # check Wav2Vec2CTCTokenizerOutput keys for word
           self.assertEqual(len(outputs.keys()				)		,     4				)
           self.assertTrue("""text""" in outputs				)
           self.assertTrue("""word_offsets""" in outputs				)
           self.assertTrue(isinstance(_a		,     _a				)				)

           self.assertListEqual(
               [""" """.join(self.get_from_offsets(_a		,     """word"""				)				) for o in outputs["""word_offsets"""]]		,     outputs.text				)
           self.assertListEqual(self.get_from_offsets(outputs["""word_offsets"""][0]		,     """word"""				)		,     ["""<s>""", """<s>""", """</s>"""]				)
           self.assertListEqual(self.get_from_offsets(outputs["""word_offsets"""][0]		,     """start_offset"""				)		,     [0, 2, 4]				)
           self.assertListEqual(self.get_from_offsets(outputs["""word_offsets"""][0]		,     """end_offset"""				)		,     [1, 3, 5]				)



     @slow
     @require_torch
     @require_torchaudio
     def 					a__	(  self				)						->							Tuple:
           import torch

           _A							:     Union[str, Any]			=						load_dataset("""common_voice"""		,     """en"""		,     split="""train"""		,     streaming=_a				)
           _A							:     List[Any]			=						ds.cast_column("""audio"""		,     datasets.Audio(sampling_rate=1_6000				)				)
           _A							:     Optional[Any]			=						iter(_a				)
           _A							:     Any			=						next(_a				)

           _A							:     str			=						AutoProcessor.from_pretrained("""patrickvonplaten/wav2vec2-base-100h-with-lm"""				)
           _A							:     Optional[int]			=						WavaVecaForCTC.from_pretrained("""patrickvonplaten/wav2vec2-base-100h-with-lm"""				)

           # compare to filename `common_voice_en_100038.mp3` of dataset viewer on https://huggingface.co/datasets/common_voice/viewer/en/train
           _A							:     Dict			=						processor(sample["""audio"""]["""array"""]		,     return_tensors="""pt"""				).input_values

           with torch.no_grad():
                 _A							:     Dict			=						model(_a				).logits.cpu().numpy()

           _A							:     Optional[int]			=						processor.decode(logits[0]		,     output_word_offsets=_a				)

           _A							:     List[str]			=						model.config.inputs_to_logits_ratio / processor.feature_extractor.sampling_rate
           _A							:     List[Any]			=						[
               {
                   """start_time""": d["""start_offset"""] * time_offset,
                   """end_time""": d["""end_offset"""] * time_offset,
                   """word""": d["""word"""],
               }
               for d in output["""word_offsets"""]
           ]

           _A							:     Optional[Any]			=						"""WHY DOES MILISANDRA LOOK LIKE SHE WANTS TO CONSUME JOHN SNOW ON THE RIVER AT THE WALL"""

           # output words
           self.assertEqual(""" """.join(self.get_from_offsets(_a		,     """word"""				)				)		,     _a				)
           self.assertEqual(""" """.join(self.get_from_offsets(_a		,     """word"""				)				)		,     output.text				)

           # output times
           _A							:     Optional[Any]			=						torch.tensor(self.get_from_offsets(_a		,     """start_time"""				)				)
           _A							:     int			=						torch.tensor(self.get_from_offsets(_a		,     """end_time"""				)				)

           # fmt: off
           _A							:     Dict			=						torch.tensor([1.4199, 1.6599, 2.2599, 3.0, 3.24, 3.5999, 3.7999, 4.0999, 4.26, 4.94, 5.28, 5.6599, 5.78, 5.94, 6.32, 6.5399, 6.6599]				)
           _A							:     str			=						torch.tensor([1.5399, 1.8999, 2.9, 3.16, 3.5399, 3.72, 4.0199, 4.1799, 4.76, 5.1599, 5.5599, 5.6999, 5.86, 6.1999, 6.38, 6.6199, 6.94]				)
           # fmt: on

           self.assertTrue(torch.allclose(_a		,     _a		,     atol=0.01				)				)
           self.assertTrue(torch.allclose(_a		,     _a		,     atol=0.01				)				)




import copy
from typing import Any, Dict, List, Optional, Union

import numpy as np

from ...audio_utils import mel_filter_bank, spectrogram, window_function
from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import TensorType, logging


lowercase_									=   logging.get_logger(__name__)





class  __lowerCAmelCase     (					SCREAMING_SNAKE_CASE							):
  _a									=  ["""input_features"""]




  def __init__(				self							,   lowerCAmelCase=80							,   lowerCAmelCase=16_000							,   lowerCAmelCase=160							,   lowerCAmelCase=30							,   lowerCAmelCase=400							,   lowerCAmelCase=0.0							,   lowerCAmelCase=False							,   **lowerCAmelCase							,   )  -> Any:
        '''simple docstring'''






        super().__init__(
            feature_size=lowerCAmelCase							,   sampling_rate=lowerCAmelCase							,   padding_value=lowerCAmelCase							,   return_attention_mask=lowerCAmelCase							,   **lowerCAmelCase							,   )
        _lowercase	       =n_fft
        _lowercase	       =hop_length
        _lowercase	       =chunk_length
        _lowercase	       =chunk_length * sampling_rate
        _lowercase	       =self.n_samples // hop_length
        _lowercase	       =sampling_rate
        _lowercase	       =mel_filter_bank(
            num_frequency_bins=1 + n_fft // 2							,   num_mel_filters=lowerCAmelCase							,   min_frequency=0.0							,   max_frequency=8000.0							,   sampling_rate=lowerCAmelCase							,   norm='slaney'							,   mel_scale='slaney'							,   )




  def   A__     (				self							,   lowerCAmelCase					)  -> np.ndarray:
        '''simple docstring'''






        _lowercase	       =spectrogram(
            lowerCAmelCase							,   window_function(self.n_fft							,   'hann'					)							,   frame_length=self.n_fft							,   hop_length=self.hop_length							,   power=2.0							,   mel_filters=self.mel_filters							,   log_mel='log10'							,   )
        _lowercase	       =log_spec[:, :-1]
        _lowercase	       =np.maximum(lowerCAmelCase							,   log_spec.max() - 8.0					)
        _lowercase	       =(log_spec + 4.0) / 4.0
        return log_spec




  @staticmethod
  # Copied from transformers.models.wav2vec2.feature_extraction_wav2vec2.Wav2Vec2FeatureExtractor.zero_mean_unit_var_norm
  def   A__     (				lowerCAmelCase							,   lowerCAmelCase							,   lowerCAmelCase = 0.0					)  -> List[np.ndarray]:
        '''simple docstring'''






        if attention_mask is not None:
              _lowercase	       =np.array(lowerCAmelCase							,   np.intaa					)
              _lowercase	       =[]

              for vector, length in zip(lowerCAmelCase							,   attention_mask.sum(-1					)					):
                    _lowercase	       =(vector - vector[:length].mean()) / np.sqrt(vector[:length].var() + 1e-7					)
                    if length < normed_slice.shape[0]:
                          _lowercase	       =padding_value

                    normed_input_values.append(lowerCAmelCase					)
        else:
              _lowercase	       =[(x - x.mean()) / np.sqrt(x.var() + 1e-7					) for x in input_values]

        return normed_input_values




  def __call__(				self							,   lowerCAmelCase							,   lowerCAmelCase = True							,   lowerCAmelCase = None							,   lowerCAmelCase = None							,   lowerCAmelCase = None							,   lowerCAmelCase = "max_length"							,   lowerCAmelCase = None							,   lowerCAmelCase = None							,   lowerCAmelCase = None							,   **lowerCAmelCase							,   )  -> BatchFeature:
        '''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.__class__.__name__} was trained using a'''
                        F''' sampling rate of {self.sampling_rate}. Please make sure that the provided `raw_speech` input'''
                        F''' was sampled with {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.'					)

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

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

        # always return batch
        if not is_batched:
              _lowercase	       =[np.asarray([raw_speech]					).T]

        _lowercase	       =BatchFeature({'input_features': raw_speech}					)

        # convert into correct format for padding

        _lowercase	       =self.pad(
            lowerCAmelCase							,   padding=lowerCAmelCase							,   max_length=max_length if max_length else self.n_samples							,   truncation=lowerCAmelCase							,   pad_to_multiple_of=lowerCAmelCase							,   return_attention_mask=return_attention_mask or do_normalize							,   )

        # zero-mean and unit-variance normalization
        if do_normalize:
              _lowercase	       =self.zero_mean_unit_var_norm(
                  padded_inputs['input_features']							,   attention_mask=padded_inputs['attention_mask']							,   padding_value=self.padding_value							,   )
              _lowercase	       =np.stack(padded_inputs['input_features']							,   axis=0					)

        # make sure list is in array format
        _lowercase	       =padded_inputs.get('input_features'					).transpose(2							,   0							,   1					)

        _lowercase	       =[self._np_extract_fbank_features(lowerCAmelCase					) for waveform in input_features[0]]

        if isinstance(input_features[0]							,   lowerCAmelCase					):
              _lowercase	       =[np.asarray(lowerCAmelCase							,   dtype=np.floataa					) for feature in input_features]
        else:
              _lowercase	       =input_features

        if return_attention_mask:
              # rescale from sample (48000) to feature (3000)
              _lowercase	       =padded_inputs['attention_mask'][:, :: self.hop_length]

        if return_tensors is not None:
              _lowercase	       =padded_inputs.convert_to_tensors(lowerCAmelCase					)

        return padded_inputs




  def   A__     (				self					)  -> Dict[str, Any]:
        '''simple docstring'''






        _lowercase	       =copy.deepcopy(self.__dict__					)
        _lowercase	       =self.__class__.__name__
        if "mel_filters" in output:
              del output["mel_filters"]
        return output


