



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


lowercase  							=					logging.get_logger(__name__)




class 			lowercase__			(      A    ):
							'''simple docstring'''







							_UpperCAmelCase      				=					['''input_features''', '''attention_mask''']






							def __init__(  self	,       snake_case=80	,       snake_case=16000	,       snake_case=80	,       snake_case=0.0	,       snake_case=True	,       snake_case=True	,       snake_case=True	,       **snake_case	,       )			->							Tuple:
									super().__init__(feature_size=snake_case	,       sampling_rate=snake_case	,       padding_value=snake_case	,       **snake_case						)
									_UpperCAmelCase							 =		num_mel_bins
									_UpperCAmelCase							 =		do_ceptral_normalize
									_UpperCAmelCase							 =		normalize_means
									_UpperCAmelCase							 =		normalize_vars
									_UpperCAmelCase							 =		True






							def        lowerCamelCase_   (  self	,       snake_case	,       )			->							np.ndarray:
									_UpperCAmelCase							 =		waveform * (2**15)  # Kaldi compliance: 16-bit signed integers
									_UpperCAmelCase							 =		torch.from_numpy(snake_case						).unsqueeze(0						)
									_UpperCAmelCase							 =		ta_kaldi.fbank(snake_case	,       num_mel_bins=self.num_mel_bins	,       sample_frequency=self.sampling_rate						)
									return features.numpy()






							@staticmethod
							def        lowerCamelCase_   (  snake_case	,       snake_case	,       snake_case = True	,       snake_case = True	,       snake_case = 0.0	,       )			->							np.ndarray:
									# make sure we normalize float32 arrays
									if normalize_means:
											_UpperCAmelCase							 =		x[:input_length].mean(axis=0						)
											_UpperCAmelCase							 =		np.subtract(snake_case	,       snake_case						)
									if normalize_vars:
											_UpperCAmelCase							 =		x[:input_length].std(axis=0						)
											_UpperCAmelCase							 =		np.divide(snake_case	,       snake_case						)

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

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

									return x






							def        lowerCamelCase_   (  self	,       snake_case	,       snake_case = None						)			->							List[np.ndarray]:
									_UpperCAmelCase							 =		attention_mask.sum(-1						) if attention_mask is not None else [x.shape[0] for x in input_features]
									return [
									    self.utterance_cmvn(snake_case	,       snake_case	,       self.normalize_means	,       self.normalize_vars	,       self.padding_value						)
									    for x, n in zip(snake_case	,       snake_case						)
									]






							def __call__(  self	,       snake_case	,       snake_case = False	,       snake_case = None	,       snake_case = False	,       snake_case = None	,       snake_case = None	,       snake_case = None	,       snake_case = None	,       **snake_case	,       )			->							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.'						)

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

									if is_batched:
											_UpperCAmelCase							 =		[np.asarray(snake_case	,       dtype=np.floataa						) for speech in raw_speech]
									elif not is_batched and not isinstance(snake_case	,       np.ndarray						):
											_UpperCAmelCase							 =		np.asarray(snake_case	,       dtype=np.floataa						)
									elif isinstance(snake_case	,       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
									_UpperCAmelCase							 =		[self._extract_fbank_features(snake_case						) for waveform in raw_speech]

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

									_UpperCAmelCase							 =		self.pad(
									    snake_case	,       padding=snake_case	,       max_length=snake_case	,       truncation=snake_case	,       pad_to_multiple_of=snake_case	,       return_attention_mask=snake_case	,       **snake_case	,       )

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

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

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

									if return_tensors is not None:
											_UpperCAmelCase							 =		padded_inputs.convert_to_tensors(snake_case						)

									return padded_inputs







"""simple docstring"""





from typing import List, Optional, TypeVar

from .arrow_dataset import Dataset, _concatenate_map_style_datasets, _interleave_map_style_datasets
from .dataset_dict import DatasetDict, IterableDatasetDict
from .info import DatasetInfo
from .iterable_dataset import IterableDataset, _concatenate_iterable_datasets, _interleave_iterable_datasets
from .splits import NamedSplit
from .utils import logging
from .utils.py_utils import Literal


lowercase  							=					logging.get_logger(__name__)


lowercase  							=					TypeVar('''DatasetType''', Dataset, IterableDataset)




def 	UpperCAmelCase     (		A	:						List[DatasetType]     ,				A	:						Optional[List[float]] = None     ,				A	:						Optional[int] = None     ,				A	:						Optional[DatasetInfo] = None     ,				A	:						Optional[NamedSplit] = None     ,				A	:						Literal["first_exhausted", "all_exhausted"] = "first_exhausted"     ,				):

		'''simple docstring'''


		from .arrow_dataset import Dataset
		from .iterable_dataset import IterableDataset

		if not datasets:
				raise ValueError('Unable to interleave an empty list of datasets.'       )
		for i, dataset in enumerate(A       ):
				if not isinstance(A     ,				(Dataset, IterableDataset)       ):
						if isinstance(A     ,				(DatasetDict, IterableDatasetDict)       ):
								if not dataset:
										raise ValueError(
										    f'Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} '
										    'is an empty dataset dictionary.'       )
								raise ValueError(
								    f'Dataset at position {i} has at least one split: {list(A       )}\n'
								    f'Please pick one to interleave with the other datasets, for example: dataset[\'{next(iter(A       )       )}\']'       )
						raise ValueError(
						    f'Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} is a {type(A       ).__name__}.'       )
				if i == 0:
						_UpperCAmelCase ,					_UpperCAmelCase							 =		(
						    (Dataset, IterableDataset) if isinstance(A     ,				A       ) else (IterableDataset, Dataset)
						)
				elif not isinstance(A     ,				A       ):
						raise ValueError(
						    f'Unable to interleave a {dataset_type.__name__} (at position 0) with a {other_type.__name__} (at position {i}). Expected a list of Dataset objects or a list of IterableDataset objects.'       )
		if stopping_strategy not in ["first_exhausted", "all_exhausted"]:
				raise ValueError(f'{stopping_strategy} is not supported. Please enter a valid stopping_strategy.'       )
		if dataset_type is Dataset:
				return _interleave_map_style_datasets(
				    A     ,				A     ,				A     ,				info=A     ,				split=A     ,				stopping_strategy=A       )
		else:
				return _interleave_iterable_datasets(
				    A     ,				A     ,				A     ,				info=A     ,				split=A     ,				stopping_strategy=A       )






def 	UpperCAmelCase     (		A	:						List[DatasetType]     ,				A	:						Optional[DatasetInfo] = None     ,				A	:						Optional[NamedSplit] = None     ,				A	:						int = 0     ,				):

		'''simple docstring'''


		if not dsets:
				raise ValueError('Unable to concatenate an empty list of datasets.'       )
		for i, dataset in enumerate(A       ):
				if not isinstance(A     ,				(Dataset, IterableDataset)       ):
						if isinstance(A     ,				(DatasetDict, IterableDatasetDict)       ):
								if not dataset:
										raise ValueError(
										    f'Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} '
										    'is an empty dataset dictionary.'       )
								raise ValueError(
								    f'Dataset at position {i} has at least one split: {list(A       )}\n'
								    f'Please pick one to interleave with the other datasets, for example: dataset[\'{next(iter(A       )       )}\']'       )
						raise ValueError(
						    f'Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} is a {type(A       ).__name__}.'       )
				if i == 0:
						_UpperCAmelCase ,					_UpperCAmelCase							 =		(
						    (Dataset, IterableDataset) if isinstance(A     ,				A       ) else (IterableDataset, Dataset)
						)
				elif not isinstance(A     ,				A       ):
						raise ValueError(
						    f'Unable to interleave a {dataset_type.__name__} (at position 0) with a {other_type.__name__} (at position {i}). Expected a list of Dataset objects or a list of IterableDataset objects.'       )
		if dataset_type is Dataset:
				return _concatenate_map_style_datasets(A     ,				info=A     ,				split=A     ,				axis=A       )
		else:
				return _concatenate_iterable_datasets(A     ,				info=A     ,				split=A     ,				axis=A       )


