





import itertools
import random
import unittest

import numpy as np

from transformers import ASTFeatureExtractor
from transformers.testing_utils import require_torch, require_torchaudio
from transformers.utils.import_utils import is_torch_available

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


UpperCAmelCase	:	Dict		     =       random.Random()

if is_torch_available():
     import torch
def      _A       (		SCREAMING_SNAKE_CASE     :   Optional[Any]       ,       SCREAMING_SNAKE_CASE     :   Any=1.0       ,       SCREAMING_SNAKE_CASE     :   Union[str, Any]=None       ,       SCREAMING_SNAKE_CASE     :   List[str]=None						):





  """simple docstring"""



  if rng is None:
    a__	:					Any										=global_rng

  a__	:					List[Any]										=[]
  for batch_idx in range(shape[0]						):
    values.append([]						)
    for _ in range(shape[1]						):
      values[-1].append(rng.random() * scale						)

  return values







class 				__lowerCAmelCase       (    unittest.TestCase):







      def __init__(    self		,      lowerCAmelCase__		,      lowerCAmelCase__=7		,      lowerCAmelCase__=4_0_0		,      lowerCAmelCase__=2_0_0_0		,      lowerCAmelCase__=1		,      lowerCAmelCase__=0.0		,      lowerCAmelCase__=1_6_0_0_0		,      lowerCAmelCase__=True		,      lowerCAmelCase__=True		,      )   ->						Any:



        '''simple docstring'''




        a__	:					List[Any]										=parent
        a__	:					Optional[int]										=batch_size
        a__	:					Optional[Any]										=min_seq_length
        a__	:					List[str]										=max_seq_length
        a__	:					Any										=(self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
        a__	:					Dict										=feature_size
        a__	:					Dict										=padding_value
        a__	:					Optional[int]										=sampling_rate
        a__	:					Optional[int]										=return_attention_mask
        a__	:					Optional[Any]										=do_normalize







      def    _lowercase  (    self							)   ->						Optional[int]:



        '''simple docstring'''




        return {
            "feature_size": self.feature_size,
            "padding_value": self.padding_value,
            "sampling_rate": self.sampling_rate,
            "return_attention_mask": self.return_attention_mask,
            "do_normalize": self.do_normalize,
        }







      def    _lowercase  (    self		,      lowerCAmelCase__=False		,      lowerCAmelCase__=False							)   ->						Any:



        '''simple docstring'''




        def _flatten(lowerCAmelCase__							):
          return list(itertools.chain(*lowerCAmelCase__							)							)

        if equal_length:
          a__	:					str										=floats_list((self.batch_size, self.max_seq_length)							)
        else:
          # make sure that inputs increase in size
          a__	:					Union[str, Any]										=[
              _flatten(floats_list((x, self.feature_size)							)							)
              for x in range(self.min_seq_length		,      self.max_seq_length		,      self.seq_length_diff							)
          ]

        if numpify:
          a__	:					List[Any]										=[np.asarray(lowerCAmelCase__							) for x in speech_inputs]

        return speech_inputs







@require_torch
@require_torchaudio
class 				__lowerCAmelCase       (    UpperCamelCase__						,				unittest.TestCase):
      _lowercase				:					int						=			ASTFeatureExtractor







      def    _lowercase  (    self							)   ->						Optional[int]:



        '''simple docstring'''




        a__	:					Optional[int]										=ASTFeatureExtractionTester(self							)







      def    _lowercase  (    self							)   ->						str:



        '''simple docstring'''




        a__	:					str										=self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()							)
        # create three inputs of length 800, 1000, and 1200
        a__	:					str										=[floats_list((1, x)							)[0] for x in range(8_0_0		,      1_4_0_0		,      2_0_0							)]
        a__	:					List[Any]										=[np.asarray(lowerCAmelCase__							) for speech_input in speech_inputs]

        # Test not batched input
        a__	:					Dict										=feat_extract(speech_inputs[0]		,      return_tensors="np"							).input_values
        a__	:					Optional[int]										=feat_extract(np_speech_inputs[0]		,      return_tensors="np"							).input_values
        self.assertTrue(np.allclose(lowerCAmelCase__		,      lowerCAmelCase__		,      atol=1E-3							)							)

        # Test batched
        a__	:					Any										=feat_extract(lowerCAmelCase__		,      padding=lowerCAmelCase__		,      return_tensors="np"							).input_values
        a__	:					str										=feat_extract(lowerCAmelCase__		,      padding=lowerCAmelCase__		,      return_tensors="np"							).input_values
        for enc_seq_a, enc_seq_a in zip(lowerCAmelCase__		,      lowerCAmelCase__							):
          self.assertTrue(np.allclose(lowerCAmelCase__		,      lowerCAmelCase__		,      atol=1E-3							)							)

        # Test 2-D numpy arrays are batched.
        a__	:					str										=[floats_list((1, x)							)[0] for x in (8_0_0, 8_0_0, 8_0_0)]
        a__	:					Dict										=np.asarray(lowerCAmelCase__							)
        a__	:					List[Any]										=feat_extract(lowerCAmelCase__		,      return_tensors="np"							).input_values
        a__	:					Optional[int]										=feat_extract(lowerCAmelCase__		,      return_tensors="np"							).input_values
        for enc_seq_a, enc_seq_a in zip(lowerCAmelCase__		,      lowerCAmelCase__							):
          self.assertTrue(np.allclose(lowerCAmelCase__		,      lowerCAmelCase__		,      atol=1E-3							)							)







      @require_torch
      def    _lowercase  (    self							)   ->						int:



        '''simple docstring'''




        import torch

        a__	:					Union[str, Any]										=self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()							)
        a__	:					Tuple										=np.random.rand(1_0_0							).astype(np.floataa							)
        a__	:					Any										=np_speech_inputs.tolist()

        for inputs in [py_speech_inputs, np_speech_inputs]:
          a__	:					int										=feature_extractor.pad([{"input_values": inputs}]		,      return_tensors="np"							)
          self.assertTrue(np_processed.input_values.dtype == np.floataa							)
          a__	:					List[str]										=feature_extractor.pad([{"input_values": inputs}]		,      return_tensors="pt"							)
          self.assertTrue(pt_processed.input_values.dtype == torch.floataa							)







      def    _lowercase  (    self		,      lowerCAmelCase__							)   ->						List[Any]:



        '''simple docstring'''




        from datasets import load_dataset

        a__	:					int										=load_dataset("hf-internal-testing/librispeech_asr_dummy"		,      "clean"		,      split="validation"							)
        # automatic decoding with librispeech
        a__	:					str										=ds.sort("id"							).select(range(lowerCAmelCase__							)							)[:num_samples]["audio"]

        return [x["array"] for x in speech_samples]







      @require_torch
      def    _lowercase  (    self							)   ->						int:



        '''simple docstring'''




        a__	:					Optional[Any]										=torch.tensor(
            [-0.98_94, -1.27_76, -0.90_66, -1.27_76, -0.93_49, -1.26_09, -1.03_86, -1.27_76,
             -1.15_61, -1.27_76, -1.20_52, -1.27_23, -1.21_90, -1.21_32, -1.27_76, -1.11_33,
             -1.19_53, -1.13_43, -1.15_84, -1.22_03, -1.17_70, -1.24_74, -1.23_81, -1.19_36,
             -0.92_70, -0.83_17, -0.80_49, -0.77_06, -0.75_65, -0.78_69]							)
        # fmt: on

        a__	:					str										=self._load_datasamples(1							)
        a__	:					Dict										=ASTFeatureExtractor()
        a__	:					str										=feature_extractor(lowerCAmelCase__		,      return_tensors="pt"							).input_values
        self.assertEquals(input_values.shape		,      (1, 1_0_2_4, 1_2_8)							)
        self.assertTrue(torch.allclose(input_values[0, 0, :3_0]		,      lowerCAmelCase__		,      atol=1E-4							)							)













from typing import Optional

import pyspark

from .. import Features, NamedSplit
from ..download import DownloadMode
from ..packaged_modules.spark.spark import Spark
from .abc import AbstractDatasetReader







class 				__lowerCAmelCase       (    UpperCamelCase__):







      def __init__(    self		,      lowerCAmelCase__		,      lowerCAmelCase__ = None		,      lowerCAmelCase__ = None		,      lowerCAmelCase__ = True		,      lowerCAmelCase__ = None		,      lowerCAmelCase__ = False		,      lowerCAmelCase__ = None		,      lowerCAmelCase__ = True		,      lowerCAmelCase__ = "arrow"		,      **lowerCAmelCase__		,      )   ->						Optional[int]:



        '''simple docstring'''




        super().__init__(
            split=lowerCAmelCase__		,      features=lowerCAmelCase__		,      cache_dir=lowerCAmelCase__		,      keep_in_memory=lowerCAmelCase__		,      streaming=lowerCAmelCase__		,      **lowerCAmelCase__		,      )
        a__	:					int										=load_from_cache_file
        a__	:					Tuple										=file_format
        a__	:					List[Any]										=Spark(
            df=lowerCAmelCase__		,      features=lowerCAmelCase__		,      cache_dir=lowerCAmelCase__		,      working_dir=lowerCAmelCase__		,      **lowerCAmelCase__		,      )







      def    _lowercase  (    self							)   ->						str:



        '''simple docstring'''




        if self.streaming:
          return self.builder.as_streaming_dataset(split=self.split							)
        a__	:					str										=None if self._load_from_cache_file else DownloadMode.FORCE_REDOWNLOAD
        self.builder.download_and_prepare(
            download_mode=lowerCAmelCase__		,      file_format=self._file_format		,      )
        return self.builder.as_dataset(split=self.split							)






