
'''simple docstring'''

from typing import List, Optional, Union

import numpy as np

from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import PaddingStrategy, TensorType, logging


__snake_case:  List[Any]								=			logging.get_logger(__name__)


class  _UpperCAmelCase ( lowerCAmelCase__    ):

     """simple docstring"""

     a_    	=					["input_values", "padding_mask"]


     def __init__(      self							,		lowerCAmelCase_ = 1							,		lowerCAmelCase_ = 2_40_00							,		lowerCAmelCase_ = 0.0							,		lowerCAmelCase_ = None							,		lowerCAmelCase_ = None							,		**lowerCAmelCase_							,		):

       '''simple docstring'''

       super().__init__(feature_size=lowerCAmelCase_							,		sampling_rate=lowerCAmelCase_							,		padding_value=lowerCAmelCase_							,		**lowerCAmelCase_ )
       a_      :	Optional[int]      =							chunk_length_s
       a_      :	Dict      =							overlap


     @property
     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       if self.chunk_length_s is None:
         return None
       else:
         return int(self.chunk_length_s * self.sampling_rate )


     @property
     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       if self.chunk_length_s is None or self.overlap is None:
         return None
       else:
         return max(1							,		int((1.0 - self.overlap) * self.chunk_length ) )






     def __call__(      self							,		lowerCAmelCase_							,		lowerCAmelCase_ = None							,		lowerCAmelCase_ = False							,		lowerCAmelCase_ = None							,		lowerCAmelCase_ = None							,		lowerCAmelCase_ = None							,		):

       '''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} was trained using a sampling rate of'''
               f''' {self.sampling_rate}. Please make sure that the provided audio 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.""" )

       if padding and truncation:
         raise ValueError("""Both padding and truncation were set. Make sure you only set one.""" )
       elif padding is None:
         # by default let's pad the inputs
         a_      :	Tuple      =							True

       a_      :	Union[str, Any]      =							bool(
           isinstance(lowerCAmelCase_							,		(list, tuple) ) and (isinstance(raw_audio[0]							,		(np.ndarray, tuple, list) )) )

       if is_batched:
         a_      :	Tuple      =							[np.asarray(lowerCAmelCase_							,		dtype=np.floataa ).T for audio in raw_audio]
       elif not is_batched and not isinstance(lowerCAmelCase_							,		np.ndarray ):
         a_      :	List[Any]      =							np.asarray(lowerCAmelCase_							,		dtype=np.floataa )
       elif isinstance(lowerCAmelCase_							,		np.ndarray ) and raw_audio.dtype is np.dtype(np.floataa ):
         a_      :	Union[str, Any]      =							raw_audio.astype(np.floataa )

       # always return batch
       if not is_batched:
         a_      :	Optional[Any]      =							[np.asarray(lowerCAmelCase_ ).T]

       # verify inputs are valid
       for idx, example in enumerate(lowerCAmelCase_ ):
         if example.ndim > 2:
           raise ValueError(f'''Expected input shape (channels, length) but got shape {example.shape}''' )
         if self.feature_size == 1 and example.ndim != 1:
           raise ValueError(f'''Expected mono audio but example has {example.shape[-1]} channels''' )
         if self.feature_size == 2 and example.shape[-1] != 2:
           raise ValueError(f'''Expected stereo audio but example has {example.shape[-1]} channels''' )

       a_      :	Optional[Any]      =							None
       a_      :	Tuple      =							BatchFeature({"""input_values""": raw_audio} )
       if self.chunk_stride is not None and self.chunk_length is not None and max_length is None:
         if truncation:
           a_      :	int      =							min(array.shape[0] for array in raw_audio )
           a_      :	Optional[int]      =							int(np.floor(max_length / self.chunk_stride ) )
           a_      :	Union[str, Any]      =							(nb_step - 1) * self.chunk_stride + self.chunk_length
         elif padding:
           a_      :	Dict      =							max(array.shape[0] for array in raw_audio )
           a_      :	List[Any]      =							int(np.ceil(max_length / self.chunk_stride ) )
           a_      :	List[Any]      =							(nb_step - 1) * self.chunk_stride + self.chunk_length
           a_      :	Union[str, Any]      =							"""max_length"""
         else:
           a_      :	int      =							input_values

        # normal padding on batch
       if padded_inputs is None:
         a_      :	int      =							self.pad(
             lowerCAmelCase_							,		max_length=lowerCAmelCase_							,		truncation=lowerCAmelCase_							,		padding=lowerCAmelCase_							,		return_attention_mask=lowerCAmelCase_							,		)
         if padding:
           a_      :	Dict      =							padded_inputs.pop("""attention_mask""" )

       a_      :	List[Any]      =							[]
       for example in padded_inputs.pop("""input_values""" ):
         if self.feature_size == 1:
           a_      :	List[str]      =							example[..., None]
         input_values.append(example.T )

       a_      :	Optional[Any]      =							input_values
       if return_tensors is not None:
         a_      :	List[Any]      =							padded_inputs.convert_to_tensors(lowerCAmelCase_ )

       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


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


__snake_case:  Tuple								=			TypeVar("DatasetType", Dataset, IterableDataset)






def 	_snake_case						(      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:
      a_      , a_      :	Union[str, Any]      =							(
          (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 	_snake_case						(      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:
      a_      , a_      :	str      =							(
          (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_   )


