


'''simple docstring'''



import multiprocessing
import os
from typing import BinaryIO, Optional, Union

import fsspec

from .. import Dataset, Features, NamedSplit, config
from ..formatting import query_table
from ..packaged_modules.json.json import Json
from ..utils import logging
from ..utils.typing import NestedDataStructureLike, PathLike
from .abc import AbstractDatasetReader





class   UpperCAmelCase_    (_UpperCAmelCase    ):





    """simple docstring"""






    def __init__(				self     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_ = None     ,	SCREAMING_SNAKE_CASE_ = None     ,	SCREAMING_SNAKE_CASE_ = None     ,	SCREAMING_SNAKE_CASE_ = False     ,	SCREAMING_SNAKE_CASE_ = False     ,	SCREAMING_SNAKE_CASE_ = None     ,	SCREAMING_SNAKE_CASE_ = None     ,	**SCREAMING_SNAKE_CASE_     ,	)							->	Dict:
         super().__init__(
             SCREAMING_SNAKE_CASE_     ,	split=SCREAMING_SNAKE_CASE_     ,	features=SCREAMING_SNAKE_CASE_     ,	cache_dir=SCREAMING_SNAKE_CASE_     ,	keep_in_memory=SCREAMING_SNAKE_CASE_     ,	streaming=SCREAMING_SNAKE_CASE_     ,	num_proc=SCREAMING_SNAKE_CASE_     ,	**SCREAMING_SNAKE_CASE_     ,	)
         __lowerCamelCase  :      Any														=						field
         __lowerCamelCase  :      Tuple														=						path_or_paths if isinstance(SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_      ) else {self.split: path_or_paths}
         __lowerCamelCase  :      int														=						Json(
             cache_dir=SCREAMING_SNAKE_CASE_     ,	data_files=SCREAMING_SNAKE_CASE_     ,	features=SCREAMING_SNAKE_CASE_     ,	field=SCREAMING_SNAKE_CASE_     ,	**SCREAMING_SNAKE_CASE_     ,	)







    def  lowercase_							(				self      )							->	List[str]:
         # Build iterable dataset
         if self.streaming:
              __lowerCamelCase  :      str														=						self.builder.as_streaming_dataset(split=self.split      )
         # Build regular (map-style) dataset
         else:
              __lowerCamelCase  :      List[Any]														=						None
              __lowerCamelCase  :      Optional[Any]														=						None
              __lowerCamelCase  :      Union[str, Any]														=						None
              __lowerCamelCase  :      Dict														=						None

              self.builder.download_and_prepare(
                  download_config=SCREAMING_SNAKE_CASE_     ,	download_mode=SCREAMING_SNAKE_CASE_     ,	verification_mode=SCREAMING_SNAKE_CASE_     ,	base_path=SCREAMING_SNAKE_CASE_     ,	num_proc=self.num_proc     ,	)
              __lowerCamelCase  :      List[str]														=						self.builder.as_dataset(
                  split=self.split     ,	verification_mode=SCREAMING_SNAKE_CASE_     ,	in_memory=self.keep_in_memory      )
         return dataset







class   UpperCAmelCase_    :





    """simple docstring"""






    def __init__(				self     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_ = None     ,	SCREAMING_SNAKE_CASE_ = None     ,	**SCREAMING_SNAKE_CASE_     ,	)							->	int:
         if num_proc is not None and num_proc <= 0:
              raise ValueError(f'num_proc {num_proc} must be an integer > 0.'      )

         __lowerCamelCase  :      Tuple														=						dataset
         __lowerCamelCase  :      Dict														=						path_or_buf
         __lowerCamelCase  :      int														=						batch_size if batch_size else config.DEFAULT_MAX_BATCH_SIZE
         __lowerCamelCase  :      str														=						num_proc
         __lowerCamelCase  :      Union[str, Any]														=						'utf-8'
         __lowerCamelCase  :      str														=						to_json_kwargs







    def  lowercase_							(				self      )							->	int:
         __lowerCamelCase  :      Optional[Any]														=						self.to_json_kwargs.pop('path_or_buf'     ,	SCREAMING_SNAKE_CASE_      )
         __lowerCamelCase  :      Tuple														=						self.to_json_kwargs.pop('orient'     ,	'records'      )
         __lowerCamelCase  :      List[Any]														=						self.to_json_kwargs.pop('lines'     ,	True if orient == 'records' else False      )
         __lowerCamelCase  :      List[str]														=						self.to_json_kwargs.pop('index'     ,	False if orient in ['split', 'table'] else True      )
         __lowerCamelCase  :      Tuple														=						self.to_json_kwargs.pop('compression'     ,	SCREAMING_SNAKE_CASE_      )

         if compression not in [None, "infer", "gzip", "bz2", "xz"]:
              raise NotImplementedError(f'`datasets` currently does not support {compression} compression'      )

         if isinstance(self.path_or_buf     ,	(str, bytes, os.PathLike)      ):
              with fsspec.open(self.path_or_buf     ,	'wb'     ,	compression=SCREAMING_SNAKE_CASE_      ) as buffer:
                   __lowerCamelCase  :      List[Any]														=						self._write(file_obj=SCREAMING_SNAKE_CASE_     ,	orient=SCREAMING_SNAKE_CASE_     ,	lines=SCREAMING_SNAKE_CASE_     ,	index=SCREAMING_SNAKE_CASE_     ,	**self.to_json_kwargs      )
         else:
              if compression:
                   raise NotImplementedError(
                       f'The compression parameter is not supported when writing to a buffer, but compression={compression}'
                       ' was passed. Please provide a local path instead.'      )
              __lowerCamelCase  :      Dict														=						self._write(
                  file_obj=self.path_or_buf     ,	orient=SCREAMING_SNAKE_CASE_     ,	lines=SCREAMING_SNAKE_CASE_     ,	index=SCREAMING_SNAKE_CASE_     ,	**self.to_json_kwargs      )
         return written







    def  lowercase_							(				self     ,	SCREAMING_SNAKE_CASE_      )							->	List[str]:
         __lowerCamelCase							,			__lowerCamelCase							,			__lowerCamelCase							,			__lowerCamelCase							,			__lowerCamelCase  :      str														=						args

         __lowerCamelCase  :      List[str]														=						query_table(
             table=self.dataset.data     ,	key=slice(SCREAMING_SNAKE_CASE_     ,	offset + self.batch_size      )     ,	indices=self.dataset._indices     ,	)
         __lowerCamelCase  :      Tuple														=						batch.to_pandas().to_json(
             path_or_buf=SCREAMING_SNAKE_CASE_     ,	orient=SCREAMING_SNAKE_CASE_     ,	lines=SCREAMING_SNAKE_CASE_     ,	index=SCREAMING_SNAKE_CASE_     ,	**SCREAMING_SNAKE_CASE_      )
         if not json_str.endswith('\n'      ):
              json_str += "\n"
         return json_str.encode(self.encoding      )







    def  lowercase_							(				self     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_     ,	**SCREAMING_SNAKE_CASE_     ,	)							->	int:
         __lowerCamelCase  :      Any														=						0

         if self.num_proc is None or self.num_proc == 1:
              for offset in logging.tqdm(
                  range(0     ,	len(self.dataset      )     ,	self.batch_size      )     ,	unit='ba'     ,	disable=not logging.is_progress_bar_enabled()     ,	desc='Creating json from Arrow format'     ,	):
                   __lowerCamelCase  :      Optional[int]														=						self._batch_json((offset, orient, lines, index, to_json_kwargs)      )
                   written += file_obj.write(SCREAMING_SNAKE_CASE_      )
         else:
              __lowerCamelCase							,			__lowerCamelCase  :      Optional[Any]														=						len(self.dataset      ), self.batch_size
              with multiprocessing.Pool(self.num_proc      ) as pool:
                   for json_str in logging.tqdm(
                       pool.imap(
                           self._batch_json     ,	[(offset, orient, lines, index, to_json_kwargs) for offset in range(0     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_      )]     ,	)     ,	total=(num_rows // batch_size) + 1 if num_rows % batch_size else num_rows // batch_size     ,	unit='ba'     ,	disable=not logging.is_progress_bar_enabled()     ,	desc='Creating json from Arrow format'     ,	):
                        written += file_obj.write(SCREAMING_SNAKE_CASE_      )

         return written













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

import numpy as np

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


SCREAMING_SNAKE_CASE__							:      Optional[int]         =    logging.get_logger(__name__)







class 	__lowerCAmelCase      (       _UpperCamelCase	):
   _UpperCamelCase			:    Tuple  		=					["""input_values""", """attention_mask"""]





   def __init__(			self       ,      snake_case = 1       ,      snake_case = 16_000       ,      snake_case = 0.0       ,      snake_case = False       ,      snake_case = 80       ,      snake_case = 16       ,      snake_case = 64       ,      snake_case = "hann_window"       ,      snake_case = 1.0       ,      snake_case = 80       ,      snake_case = 7_600       ,      snake_case = 1E-10       ,      snake_case = 2       ,      snake_case = True       ,      **snake_case       ,      )					->	Dict:
      """simple docstring"""





      super().__init__(feature_size=snake_case       ,      sampling_rate=snake_case       ,      padding_value=snake_case       ,      **snake_case   )
      a__						:    Any   		=	do_normalize
      a__						:    List[str]   		=	return_attention_mask

      a__						:    List[Any]   		=	num_mel_bins
      a__						:    List[str]   		=	hop_length
      a__						:    int   		=	win_length
      a__						:    List[Any]   		=	win_function
      a__						:    List[str]   		=	frame_signal_scale
      a__						:    List[Any]   		=	fmin
      a__						:    Optional[Any]   		=	fmax
      a__						:    Union[str, Any]   		=	mel_floor
      a__						:    Union[str, Any]   		=	reduction_factor

      a__						:    List[str]   		=	win_length * sampling_rate // 1_000
      a__						:    List[Any]   		=	hop_length * sampling_rate // 1_000
      a__						:    List[Any]   		=	optimal_fft_length(self.sample_size   )
      a__						:    Dict   		=	(self.n_fft // 2) + 1

      a__						:    str   		=	window_function(window_length=self.sample_size       ,      name=self.win_function       ,      periodic=snake_case   )

      a__						:    Tuple   		=	mel_filter_bank(
          num_frequency_bins=self.n_freqs       ,      num_mel_filters=self.num_mel_bins       ,      min_frequency=self.fmin       ,      max_frequency=self.fmax       ,      sampling_rate=self.sampling_rate       ,      norm="slaney"       ,      mel_scale="slaney"       ,      )

      if frame_signal_scale != 1.0:
         warnings.warn(
             "The argument `frame_signal_scale` is deprecated and will be removed in version 4.30.0 of Transformers"       ,      snake_case       ,      )
      if reduction_factor != 2.0:
         warnings.warn(
             "The argument `reduction_factor` is deprecated and will be removed in version 4.30.0 of Transformers"       ,      snake_case       ,      )





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





      if attention_mask is not None:
         a__						:    Tuple   		=	np.array(snake_case       ,      np.intaa   )
         a__						:    List[str]   		=	[]

         for vector, length in zip(snake_case       ,      attention_mask.sum(-1   )   ):
            a__						:    List[Any]   		=	(vector - vector[:length].mean()) / np.sqrt(vector[:length].var() + 1E-7   )
            if length < normed_slice.shape[0]:
               a__						:    List[str]   		=	padding_value

            normed_input_values.append(snake_case   )
      else:
         a__						:    Any   		=	[(x - x.mean()) / np.sqrt(x.var() + 1E-7   ) for x in input_values]

      return normed_input_values





   def 						_snake_case			(			self       ,      snake_case       ,      )					->	np.ndarray:
      """simple docstring"""





      a__						:    str   		=	spectrogram(
          snake_case       ,      window=self.window       ,      frame_length=self.sample_size       ,      hop_length=self.sample_stride       ,      fft_length=self.n_fft       ,      mel_filters=self.mel_filters       ,      mel_floor=self.mel_floor       ,      log_mel="log10"       ,      )
      return log_mel_spec.T





   def __call__(			self       ,      snake_case = None       ,      snake_case = None       ,      snake_case = False       ,      snake_case = None       ,      snake_case = False       ,      snake_case = None       ,      snake_case = None       ,      snake_case = None       ,      snake_case = None       ,      **snake_case       ,      )					->	BatchFeature:
      """simple docstring"""





      if audio is None and audio_target is None:
         raise ValueError("You must provide either `audio` or `audio_target` values."   )

      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 audio is not None:
         a__						:    Dict   		=	self._process_audio(
             snake_case       ,      snake_case       ,      snake_case       ,      snake_case       ,      snake_case       ,      snake_case       ,      snake_case       ,      snake_case       ,      **snake_case       ,      )
      else:
         a__						:    Optional[int]   		=	None

      if audio_target is not None:
         a__						:    List[Any]   		=	self._process_audio(
             snake_case       ,      snake_case       ,      snake_case       ,      snake_case       ,      snake_case       ,      snake_case       ,      snake_case       ,      snake_case       ,      **snake_case       ,      )

         if inputs is None:
            return inputs_target
         else:
            a__						:    Tuple   		=	inputs_target["input_values"]
            a__						:    Tuple   		=	inputs_target.get("attention_mask"   )
            if decoder_attention_mask is not None:
               a__						:    Tuple   		=	decoder_attention_mask

      return inputs





   def 						_snake_case			(			self       ,      snake_case       ,      snake_case = False       ,      snake_case = False       ,      snake_case = None       ,      snake_case = False       ,      snake_case = None       ,      snake_case = None       ,      snake_case = None       ,      **snake_case       ,      )					->	BatchFeature:
      """simple docstring"""





      a__						:    Optional[int]   		=	isinstance(snake_case       ,      np.ndarray   ) and len(speech.shape   ) > 1
      if is_batched_numpy and len(speech.shape   ) > 2:
         raise ValueError(F"""Only mono-channel audio is supported for input to {self}"""   )
      a__						:    List[Any]   		=	is_batched_numpy or (
          isinstance(snake_case       ,      (list, tuple)   ) and (isinstance(speech[0]       ,      (np.ndarray, tuple, list)   ))
      )

      if is_batched:
         a__						:    int   		=	[np.asarray(snake_case       ,      dtype=np.floataa   ) for speech in speech]
      elif not is_batched and not isinstance(snake_case       ,      np.ndarray   ):
         a__						:    Any   		=	np.asarray(snake_case       ,      dtype=np.floataa   )
      elif isinstance(snake_case       ,      np.ndarray   ) and speech.dtype is np.dtype(np.floataa   ):
         a__						:    List[Any]   		=	speech.astype(np.floataa   )

      # always return batch
      if not is_batched:
         a__						:    Union[str, Any]   		=	[speech]

      # needed to make pad() work on spectrogram inputs
      a__						:    Optional[Any]   		=	self.feature_size

      # convert into correct format for padding
      if is_target:
         a__						:    List[str]   		=	[self._extract_mel_features(snake_case   ) for waveform in speech]
         a__						:    Optional[Any]   		=	BatchFeature({"input_values": features}   )
         a__						:    str   		=	self.num_mel_bins
      else:
         a__						:    int   		=	BatchFeature({"input_values": speech}   )

      a__						:    int   		=	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       ,      )

      a__						:    Any   		=	feature_size_hack

      # convert input values to correct format
      a__						:    Tuple   		=	padded_inputs["input_values"]
      if not isinstance(input_values[0]       ,      np.ndarray   ):
         a__						:    int   		=	[np.asarray(snake_case       ,      dtype=np.floataa   ) for array in input_values]
      elif (
          not isinstance(snake_case       ,      np.ndarray   )
          and isinstance(input_values[0]       ,      np.ndarray   )
          and input_values[0].dtype is np.dtype(np.floataa   )
      ):
         a__						:    Union[str, Any]   		=	[array.astype(np.floataa   ) for array in input_values]
      elif isinstance(snake_case       ,      np.ndarray   ) and input_values.dtype is np.dtype(np.floataa   ):
         a__						:    Optional[int]   		=	input_values.astype(np.floataa   )

      # convert attention_mask to correct format
      a__						:    Optional[int]   		=	padded_inputs.get("attention_mask"   )
      if attention_mask is not None:
         a__						:    Tuple   		=	[np.asarray(snake_case       ,      dtype=np.intaa   ) for array in attention_mask]

      # zero-mean and unit-variance normalization
      if not is_target and self.do_normalize:
         a__						:    Any   		=	(
             attention_mask
             if self._get_padding_strategies(snake_case       ,      max_length=snake_case   ) is not PaddingStrategy.DO_NOT_PAD
             else None
         )
         a__						:    Optional[Any]   		=	self.zero_mean_unit_var_norm(
             padded_inputs["input_values"]       ,      attention_mask=snake_case       ,      padding_value=self.padding_value   )

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

      return padded_inputs





   def 						_snake_case			(			self   )					->	Dict[str, Any]:
      """simple docstring"""





      a__						:    int   		=	super().to_dict()

      # Don't serialize these as they are derived from the other properties.
      a__						:    str   		=	["window", "mel_filters", "sample_size", "sample_stride", "n_fft", "n_freqs"]
      for name in names:
         if name in output:
            del output[name]

      return output

