


'''simple docstring'''



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

import numpy as np
import torch

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      :      Union[str, Any]            =					logging.get_logger(__name__)






class 	UpperCamelCase__(       __lowercase	):
    __magic_name__    :				str  		=	["input_features", "is_longer"]




    def __init__(		self      :			str				,   lowerCAmelCase      :			List[str]=64				,   lowerCAmelCase      :			Any=48000				,   lowerCAmelCase      :			Dict=480				,   lowerCAmelCase      :			List[Any]=10				,   lowerCAmelCase      :			Optional[Any]=1024				,   lowerCAmelCase      :			Tuple=0.0				,   lowerCAmelCase      :			Any=False				,   lowerCAmelCase      :			float = 0				,   lowerCAmelCase      :			float = 14000				,   lowerCAmelCase      :			int = None				,   lowerCAmelCase      :			str = "fusion"				,   lowerCAmelCase      :			str = "repeatpad"				,   **lowerCAmelCase      :			int				,   )->					Optional[Any]:

          """simple docstring"""




          super().__init__(
              feature_size=snake_case_				,   sampling_rate=snake_case_				,   padding_value=snake_case_				,   return_attention_mask=snake_case_				,   **snake_case_				,   )
          UpperCAmelCase					    =     top_db
          UpperCAmelCase					    =     truncation
          UpperCAmelCase					    =     padding
          UpperCAmelCase					    =     fft_window_size
          UpperCAmelCase					    =     (fft_window_size >> 1) + 1
          UpperCAmelCase					    =     hop_length
          UpperCAmelCase					    =     max_length_s
          UpperCAmelCase					    =     max_length_s * sampling_rate
          UpperCAmelCase					    =     sampling_rate
          UpperCAmelCase					    =     frequency_min
          UpperCAmelCase					    =     frequency_max
          UpperCAmelCase					    =     mel_filter_bank(
              num_frequency_bins=self.nb_frequency_bins				,   num_mel_filters=snake_case_				,   min_frequency=snake_case_				,   max_frequency=snake_case_				,   sampling_rate=snake_case_				,   norm=snake_case_				,   mel_scale='''htk'''				,   )
          UpperCAmelCase					    =     mel_filter_bank(
              num_frequency_bins=self.nb_frequency_bins				,   num_mel_filters=snake_case_				,   min_frequency=snake_case_				,   max_frequency=snake_case_				,   sampling_rate=snake_case_				,   norm='''slaney'''				,   mel_scale='''slaney'''				,   )




    def 					a__(		self      :			Optional[Any]				)->					Tuple:

          """simple docstring"""




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




    def 					a__(		self      :			Tuple				,   lowerCAmelCase      :			np.array				,   lowerCAmelCase      :			Optional[np.array] = None				)->					Optional[int]:

          """simple docstring"""




          UpperCAmelCase					    =     spectrogram(
              snake_case_				,   window_function(self.fft_window_size				,   '''hann'''				)				,   frame_length=self.fft_window_size				,   hop_length=self.hop_length				,   power=2.0				,   mel_filters=snake_case_				,   log_mel='''dB'''				,   )
          return log_mel_spectrogram.T




    def 					a__(		self      :			Optional[int]				,   lowerCAmelCase      :			Any				,   lowerCAmelCase      :			Union[str, Any]				,   lowerCAmelCase      :			List[str]				)->					List[Any]:

          """simple docstring"""




          UpperCAmelCase					    =     np.array_split(list(range(0				,   total_frames - chunk_frames + 1				)				)				,   3				)
          if len(ranges[1]				) == 0:
                # if the audio is too short, we just use the first chunk
                UpperCAmelCase					    =     [0]
          if len(ranges[2]				) == 0:
                # if the audio is too short, we just use the first chunk
                UpperCAmelCase					    =     [0]
          # randomly choose index for each part
          UpperCAmelCase					    =     np.random.choice(ranges[0]				)
          UpperCAmelCase					    =     np.random.choice(ranges[1]				)
          UpperCAmelCase					    =     np.random.choice(ranges[2]				)

          UpperCAmelCase					    =     mel[idx_front : idx_front + chunk_frames, :]
          UpperCAmelCase					    =     mel[idx_middle : idx_middle + chunk_frames, :]
          UpperCAmelCase					    =     mel[idx_back : idx_back + chunk_frames, :]

          UpperCAmelCase					    =     torch.tensor(mel[None, None, :]				)
          UpperCAmelCase					    =     torch.nn.functional.interpolate(
              snake_case_				,   size=[chunk_frames, 64]				,   mode='''bilinear'''				,   align_corners=snake_case_				)
          UpperCAmelCase					    =     mel_shrink[0][0].numpy()
          UpperCAmelCase					    =     np.stack([mel_shrink, mel_chunk_front, mel_chunk_middle, mel_chunk_back]				,   axis=0				)
          return mel_fusion




    def 					a__(		self      :			Optional[Any]				,   lowerCAmelCase      :			np.array				,   lowerCAmelCase      :			Tuple				,   lowerCAmelCase      :			List[Any]				,   lowerCAmelCase      :			Optional[int]				)->					str:

          """simple docstring"""




          if waveform.shape[0] > max_length:
                if truncation == "rand_trunc":
                      UpperCAmelCase					    =     True
                      # random crop to max_length (for compatibility) -> this should be handled by self.pad
                      UpperCAmelCase					    =     len(snake_case_				) - max_length
                      UpperCAmelCase					    =     np.random.randint(0				,   overflow + 1				)
                      UpperCAmelCase					    =     waveform[idx : idx + max_length]
                      UpperCAmelCase					    =     self._np_extract_fbank_features(snake_case_				,   self.mel_filters_slaney				)[None, :]
                elif truncation == "fusion":
                      UpperCAmelCase					    =     self._np_extract_fbank_features(snake_case_				,   self.mel_filters				)
                      UpperCAmelCase					    =     max_length // self.hop_length + 1  # the +1 related to how the spectrogram is computed
                      UpperCAmelCase					    =     mel.shape[0]
                      if chunk_frames == total_frames:
                            # there is a corner case where the audio length is larger than max_length but smaller than max_length+hop_length.
                            # In this case, we just use the whole audio.
                            UpperCAmelCase					    =     np.stack([mel, mel, mel, mel]				,   axis=0				)
                            UpperCAmelCase					    =     False
                      else:
                            UpperCAmelCase					    =     self._random_mel_fusion(snake_case_				,   snake_case_				,   snake_case_				)
                            UpperCAmelCase					    =     True
                else:
                      raise NotImplementedError(F"""data_truncating {truncation} not implemented"""				)

          else:
                UpperCAmelCase					    =     False
                # only use repeat as a new possible value for padding. you repeat the audio before applying the usual max_length padding
                if waveform.shape[0] < max_length:
                      if padding == "repeat":
                            UpperCAmelCase					    =     int(max_length / len(snake_case_				)				)
                            UpperCAmelCase					    =     np.stack(np.tile(snake_case_				,   n_repeat + 1				)				)[:max_length]
                      if padding == "repeatpad":
                            UpperCAmelCase					    =     int(max_length / len(snake_case_				)				)
                            UpperCAmelCase					    =     np.stack(np.tile(snake_case_				,   snake_case_				)				)
                      UpperCAmelCase					    =     np.pad(snake_case_				,   (0, max_length - waveform.shape[0])				,   mode='''constant'''				,   constant_values=0				)

                if truncation == "fusion":
                      UpperCAmelCase					    =     self._np_extract_fbank_features(snake_case_				,   self.mel_filters				)
                      UpperCAmelCase					    =     np.stack([input_mel, input_mel, input_mel, input_mel]				,   axis=0				)
                else:
                      UpperCAmelCase					    =     self._np_extract_fbank_features(snake_case_				,   self.mel_filters_slaney				)[None, :]

          return input_mel, longer





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

          """simple docstring"""




          UpperCAmelCase					    =     truncation if truncation is not None else self.truncation
          UpperCAmelCase					    =     padding if padding else self.padding

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

          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					    =     [np.asarray(snake_case_				)]

          # convert to mel spectrogram, truncate and pad if needed.
          UpperCAmelCase					    =     [
              self._get_input_mel(snake_case_				,   max_length if max_length else self.nb_max_samples				,   snake_case_				,   snake_case_				)
              for waveform in raw_speech
          ]

          UpperCAmelCase					    =     []
          UpperCAmelCase					    =     []
          for mel, longer in padded_inputs:
                input_mel.append(snake_case_				)
                is_longer.append(snake_case_				)

          if truncation == "fusion" and sum(snake_case_				) == 0:
                # if no audio is longer than 10s, then randomly select one audio to be longer
                UpperCAmelCase					    =     np.random.randint(0				,   len(snake_case_				)				)
                UpperCAmelCase					    =     True

          if isinstance(input_mel[0]				,   snake_case_				):
                UpperCAmelCase					    =     [np.asarray(snake_case_				,   dtype=np.floataa				) for feature in input_mel]

          # is_longer is a list of bool
          UpperCAmelCase					    =     [[longer] for longer in is_longer]

          UpperCAmelCase					    =     {'''input_features''': input_mel, '''is_longer''': is_longer}
          UpperCAmelCase					    =     BatchFeature(snake_case_				)

          if return_tensors is not None:
                UpperCAmelCase					    =     input_features.convert_to_tensors(snake_case_				)

          return input_features









'''simple docstring'''



from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any, ClassVar, Dict, List, Optional, Union

import pyarrow as pa


if TYPE_CHECKING:
	from .features import FeatureType






@dataclass
class 	UpperCamelCase__:
				__magic_name__    :				List[str]
				__magic_name__    :				Optional[str]  		=	None
				# Automatically constructed
				__magic_name__    :				ClassVar[str]  		=	"dict"
				__magic_name__    :				ClassVar[Any]  		=	None
				__magic_name__    :				str  		=	field(default="Translation"						,    init=lowerCAmelCase						,    repr=lowerCAmelCase	)




				def __call__(		self      :			Union[str, Any]				)->					str:

										"""simple docstring"""




										return pa.struct({lang: pa.string() for lang in sorted(self.languages				)}				)





				def 					a__(		self      :			int				)->					Union["FeatureType", Dict[str, "FeatureType"]]:

										"""simple docstring"""




										from .features import Value

										return {k: Value('''string'''				) for k in sorted(self.languages				)}







@dataclass
class 	UpperCamelCase__:
				__magic_name__    :				Optional[List]  		=	None
				__magic_name__    :				Optional[int]  		=	None
				__magic_name__    :				Optional[str]  		=	None
				# Automatically constructed
				__magic_name__    :				ClassVar[str]  		=	"dict"
				__magic_name__    :				ClassVar[Any]  		=	None
				__magic_name__    :				str  		=	field(default="TranslationVariableLanguages"						,    init=lowerCAmelCase						,    repr=lowerCAmelCase	)




				def 					a__(		self      :			Union[str, Any]				)->					Optional[Any]:

										"""simple docstring"""




										UpperCAmelCase					    =     sorted(set(self.languages				)				) if self.languages else None
										UpperCAmelCase					    =     len(self.languages				) if self.languages else None




				def __call__(		self      :			int				)->					Optional[Any]:

										"""simple docstring"""




										return pa.struct({'''language''': pa.list_(pa.string()				), '''translation''': pa.list_(pa.string()				)}				)




				def 					a__(		self      :			Optional[int]				,   lowerCAmelCase      :			Dict				)->					Tuple:

										"""simple docstring"""




										UpperCAmelCase					    =     set(self.languages				)
										if self.languages and set(lowerCAmelCase				) - lang_set:
																raise ValueError(
																    F"""Some languages in example ({", ".join(sorted(set(lowerCAmelCase				) - lang_set				)				)}) are not in valid set ({", ".join(lowerCAmelCase				)})."""				)

										# Convert dictionary into tuples, splitting out cases where there are
										# multiple translations for a single language.
										UpperCAmelCase					    =     []
										for lang, text in translation_dict.items():
																if isinstance(lowerCAmelCase				,   lowerCAmelCase				):
																						translation_tuples.append((lang, text)				)
																else:
																						translation_tuples.extend([(lang, el) for el in text]				)

        # Ensure translations are in ascending order by language code.
										UpperCAmelCase    ,    UpperCAmelCase					    =     zip(*sorted(lowerCAmelCase				)				)

										return {"language": languages, "translation": translations}





				def 					a__(		self      :			Any				)->					Union["FeatureType", Dict[str, "FeatureType"]]:

										"""simple docstring"""




										from .features import Sequence, Value

										return {
										    "language": Sequence(Value('''string'''				)				),
										    "translation": Sequence(Value('''string'''				)				),
										}





