





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


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





class A_   (     _a  ):
   lowerCAmelCase__  	=					["""input_values""", """attention_mask"""]




   def __init__(self      :List[Any]       ,  _UpperCamelCase      :Union[str, Any]=1       ,  _UpperCamelCase      :Tuple=1_6000       ,  _UpperCamelCase      :Optional[Any]=128       ,  _UpperCamelCase      :Optional[Any]=1024       ,  _UpperCamelCase      :str=0.0       ,  _UpperCamelCase      :Tuple=True       ,  _UpperCamelCase      :Optional[Any]=-4.2_6_7_7_3_9_3       ,  _UpperCamelCase      :str=4.5_6_8_9_9_7_4       ,  _UpperCamelCase      :Optional[int]=False       ,  **_UpperCamelCase      :Union[str, Any]       ,  )->   int:
         super().__init__(feature_size=_SCREAMING_SNAKE_CASE       ,  sampling_rate=_SCREAMING_SNAKE_CASE       ,  padding_value=_SCREAMING_SNAKE_CASE       ,  **_SCREAMING_SNAKE_CASE					)
         __A			  =							num_mel_bins
         __A			  =							max_length
         __A			  =							do_normalize
         __A			  =							mean
         __A			  =							std
         __A			  =							return_attention_mask




   def _lowerCAmelCase (self      :List[str]       ,  _UpperCamelCase      :np.ndarray       ,  _UpperCamelCase      :int       ,  )->   np.ndarray:
         __A			  =							torch.from_numpy(_SCREAMING_SNAKE_CASE					).unsqueeze(0					)
         __A			  =							ta_kaldi.fbank(
             _SCREAMING_SNAKE_CASE       ,  htk_compat=_SCREAMING_SNAKE_CASE       ,  sample_frequency=self.sampling_rate       ,  use_energy=_SCREAMING_SNAKE_CASE       ,  window_type='''hanning'''       ,  num_mel_bins=self.num_mel_bins       ,  dither=0.0       ,  frame_shift=10       ,  )

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

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

         __A			  =							fbank.numpy()

         return fbank




   def _lowerCAmelCase (self      :Tuple       ,  _UpperCamelCase      :np.ndarray					)->   np.ndarray:
         return (input_values - (self.mean)) / (self.std * 2)






   def __call__(self      :Dict       ,  _UpperCamelCase      :Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]       ,  _UpperCamelCase      :Optional[int] = None       ,  _UpperCamelCase      :Optional[Union[str, TensorType]] = None       ,  **_UpperCamelCase      :Optional[Any]       ,  )->   BatchFeature:






         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(_SCREAMING_SNAKE_CASE       ,  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(_SCREAMING_SNAKE_CASE       ,  (list, tuple)					) and (isinstance(raw_speech[0]       ,  (np.ndarray, tuple, list)					))
         )

         if is_batched:
               __A			  =							[np.asarray(_SCREAMING_SNAKE_CASE       ,  dtype=np.floataa					) for speech in raw_speech]
         elif not is_batched and not isinstance(_SCREAMING_SNAKE_CASE       ,  np.ndarray					):
               __A			  =							np.asarray(_SCREAMING_SNAKE_CASE       ,  dtype=np.floataa					)
         elif isinstance(_SCREAMING_SNAKE_CASE       ,  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 and pad/truncate to max_length
         __A			  =							[self._extract_fbank_features(_SCREAMING_SNAKE_CASE       ,  max_length=self.max_length					) for waveform in raw_speech]

         # convert into BatchFeature
         __A			  =							BatchFeature({'''input_values''': features}					)

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

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

         if return_tensors is not None:
               __A			  =							padded_inputs.convert_to_tensors(_SCREAMING_SNAKE_CASE					)

         return padded_inputs







'''simple docstring'''


from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_tokenizers_available, is_torch_available


__SCREAMING_SNAKE_CASE						=      {
    "configuration_markuplm": ["MARKUPLM_PRETRAINED_CONFIG_ARCHIVE_MAP", "MarkupLMConfig"],
    "feature_extraction_markuplm": ["MarkupLMFeatureExtractor"],
    "processing_markuplm": ["MarkupLMProcessor"],
    "tokenization_markuplm": ["MarkupLMTokenizer"],
}

try:
	if not is_tokenizers_available():
		raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
	pass
else:
	__SCREAMING_SNAKE_CASE						=      ["MarkupLMTokenizerFast"]

try:
	if not is_torch_available():
		raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
	pass
else:
	__SCREAMING_SNAKE_CASE						=      [
	    "MARKUPLM_PRETRAINED_MODEL_ARCHIVE_LIST",
	    "MarkupLMForQuestionAnswering",
	    "MarkupLMForSequenceClassification",
	    "MarkupLMForTokenClassification",
	    "MarkupLMModel",
	    "MarkupLMPreTrainedModel",
	]


if TYPE_CHECKING:
	from .configuration_markuplm import MARKUPLM_PRETRAINED_CONFIG_ARCHIVE_MAP, MarkupLMConfig
	from .feature_extraction_markuplm import MarkupLMFeatureExtractor
	from .processing_markuplm import MarkupLMProcessor
	from .tokenization_markuplm import MarkupLMTokenizer

	try:
		if not is_tokenizers_available():
			raise OptionalDependencyNotAvailable()
	except OptionalDependencyNotAvailable:
		pass
	else:
		from .tokenization_markuplm_fast import MarkupLMTokenizerFast

	try:
		if not is_torch_available():
			raise OptionalDependencyNotAvailable()
	except OptionalDependencyNotAvailable:
		pass
	else:
		from .modeling_markuplm import (
		    MARKUPLM_PRETRAINED_MODEL_ARCHIVE_LIST,
		    MarkupLMForQuestionAnswering,
		    MarkupLMForSequenceClassification,
		    MarkupLMForTokenClassification,
		    MarkupLMModel,
		    MarkupLMPreTrainedModel,
		)


else:
	import sys

	__SCREAMING_SNAKE_CASE						=      _LazyModule(__name__, globals()["__file__"], _import_structure)




