





import itertools
import os
import random
import tempfile
import unittest

import numpy as np

from transformers import TvltFeatureExtractor, is_datasets_available
from transformers.testing_utils import check_json_file_has_correct_format, require_torch, require_torchaudio
from transformers.utils.import_utils import is_torch_available

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


if is_torch_available():
    import torch

if is_datasets_available():
    from datasets import load_dataset

_UpperCAmelCase       : int			     =       random.Random()


def  SCREAMING_SNAKE_CASE   ( _UpperCAmelCase		,			_UpperCAmelCase=1.0		,			_UpperCAmelCase=None		,			_UpperCAmelCase=None				)    ->  List[Any]:
  if rng is None:
    lowerCamelCase__					:   Optional[int]      				=	global_rng

  lowerCamelCase__					:   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						:   List[str]    ,  UpperCAmelCase						:   Any    ,  UpperCAmelCase						:   int=7    ,  UpperCAmelCase						:   Any=400    ,  UpperCAmelCase						:   Tuple=2000    ,  UpperCAmelCase						:   Tuple=2048    ,  UpperCAmelCase						:   List[str]=128    ,  UpperCAmelCase						:   Dict=1    ,  UpperCAmelCase						:   int=512    ,  UpperCAmelCase						:   List[str]=30    ,  UpperCAmelCase						:   List[Any]=44100    ,  )    ->			int:
   lowerCamelCase__					:   str      				=	parent
   lowerCamelCase__					:   Optional[Any]      				=	batch_size
   lowerCamelCase__					:   Optional[Any]      				=	min_seq_length
   lowerCamelCase__					:   Optional[int]      				=	max_seq_length
   lowerCamelCase__					:   List[Any]      				=	(self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
   lowerCamelCase__					:   List[str]      				=	spectrogram_length
   lowerCamelCase__					:   Union[str, Any]      				=	feature_size
   lowerCamelCase__					:   int      				=	num_audio_channels
   lowerCamelCase__					:   Tuple      				=	hop_length
   lowerCamelCase__					:   str      				=	chunk_length
   lowerCamelCase__					:   Optional[int]      				=	sampling_rate
 def   A_			(   self						:   str			)    ->			Any:
   return {
       "spectrogram_length": self.spectrogram_length,
       "feature_size": self.feature_size,
       "num_audio_channels": self.num_audio_channels,
       "hop_length": self.hop_length,
       "chunk_length": self.chunk_length,
       "sampling_rate": self.sampling_rate,
   }







 def   A_			(   self						:   int    ,  UpperCAmelCase						:   Tuple=False    ,  UpperCAmelCase						:   int=False			)    ->			Dict:
   def _flatten(UpperCAmelCase						:   Union[str, Any]			):
     return list(itertools.chain(*__A			)			)

   if equal_length:
     lowerCamelCase__					:   Any      				=	[floats_list((self.max_seq_length, self.feature_size)			) for _ in range(self.batch_size			)]
   else:
     # make sure that inputs increase in size
     lowerCamelCase__					:   Optional[Any]      				=	[
         floats_list((x, self.feature_size)			)
         for x in range(self.min_seq_length    ,  self.max_seq_length    ,  self.seq_length_diff			)
     ]
   if numpify:
     lowerCamelCase__					:   List[str]      				=	[np.asarray(__A			) for x in speech_inputs]
   return speech_inputs


@require_torch
@require_torchaudio
class      lowerCAmelCase							(      UpperCamelCase_,						unittest.TestCase    ):
 UpperCAmelCase__									=	TvltFeatureExtractor
 def   A_			(   self						:   Any			)    ->			List[str]:
   lowerCamelCase__					:   Dict      				=	TvltFeatureExtractionTester(self			)
 def   A_			(   self						:   Tuple			)    ->			Optional[Any]:
   lowerCamelCase__					:   Tuple      				=	self.feature_extraction_class(**self.feat_extract_dict			)
   self.assertTrue(hasattr(__A    ,  'spectrogram_length'			)			)
   self.assertTrue(hasattr(__A    ,  'feature_size'			)			)
   self.assertTrue(hasattr(__A    ,  'num_audio_channels'			)			)
   self.assertTrue(hasattr(__A    ,  'hop_length'			)			)
   self.assertTrue(hasattr(__A    ,  'chunk_length'			)			)
   self.assertTrue(hasattr(__A    ,  'sampling_rate'			)			)
 def   A_			(   self						:   Dict			)    ->			Union[str, Any]:
   lowerCamelCase__					:   int      				=	self.feature_extraction_class(**self.feat_extract_dict			)

   with tempfile.TemporaryDirectory() as tmpdirname:
     lowerCamelCase__					:   Optional[Any]      				=	feat_extract_first.save_pretrained(__A			)[0]
     check_json_file_has_correct_format(__A			)
     lowerCamelCase__					:   List[Any]      				=	self.feature_extraction_class.from_pretrained(__A			)

   lowerCamelCase__					:   List[str]      				=	feat_extract_first.to_dict()
   lowerCamelCase__					:   str      				=	feat_extract_second.to_dict()
   lowerCamelCase__					:   List[Any]      				=	dict_first.pop('mel_filters'			)
   lowerCamelCase__					:   Any      				=	dict_second.pop('mel_filters'			)
   self.assertTrue(np.allclose(__A    ,  __A			)			)
   self.assertEqual(__A    ,  __A			)
 def   A_			(   self						:   int			)    ->			Union[str, Any]:
   lowerCamelCase__					:   Dict      				=	self.feature_extraction_class(**self.feat_extract_dict			)

   with tempfile.TemporaryDirectory() as tmpdirname:
     lowerCamelCase__					:   List[str]      				=	os.path.join(__A    ,  'feat_extract.json'			)
     feat_extract_first.to_json_file(__A			)
     lowerCamelCase__					:   Any      				=	self.feature_extraction_class.from_json_file(__A			)

   lowerCamelCase__					:   Dict      				=	feat_extract_first.to_dict()
   lowerCamelCase__					:   Any      				=	feat_extract_second.to_dict()
   lowerCamelCase__					:   List[str]      				=	dict_first.pop('mel_filters'			)
   lowerCamelCase__					:   int      				=	dict_second.pop('mel_filters'			)
   self.assertTrue(np.allclose(__A    ,  __A			)			)
   self.assertEqual(__A    ,  __A			)
 def   A_			(   self						:   List[str]			)    ->			int:
   # Initialize feature_extractor
   lowerCamelCase__					:   Union[str, Any]      				=	self.feature_extraction_class(**self.feat_extract_dict			)

   # create three inputs of length 800, 1000, and 1200
   lowerCamelCase__					:   Dict      				=	[floats_list((1, x)			)[0] for x in range(800    ,  1400    ,  200			)]
   lowerCamelCase__					:   Union[str, Any]      				=	[np.asarray(__A			) for speech_input in speech_inputs]

   # Test not batched input
   lowerCamelCase__					:   Optional[int]      				=	feature_extractor(np_speech_inputs[0]    ,  return_tensors='np'    ,  sampling_rate=44100			).audio_values

   self.assertTrue(encoded_audios.ndim == 4			)
   self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size			)
   self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length			)
   self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels			)

   # Test batched
   lowerCamelCase__					:   Dict      				=	feature_extractor(__A    ,  return_tensors='np'    ,  sampling_rate=44100			).audio_values

   self.assertTrue(encoded_audios.ndim == 4			)
   self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size			)
   self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length			)
   self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels			)

   # Test audio masking
   lowerCamelCase__					:   Union[str, Any]      				=	feature_extractor(
       __A    ,  return_tensors='np'    ,  sampling_rate=44100    ,  mask_audio=__A			).audio_values

   self.assertTrue(encoded_audios.ndim == 4			)
   self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size			)
   self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length			)
   self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels			)

   # Test 2-D numpy arrays are batched.
   lowerCamelCase__					:   Any      				=	[floats_list((1, x)			)[0] for x in (800, 800, 800)]
   lowerCamelCase__					:   str      				=	np.asarray(__A			)
   lowerCamelCase__					:   Tuple      				=	feature_extractor(__A    ,  return_tensors='np'    ,  sampling_rate=44100			).audio_values
   self.assertTrue(encoded_audios.ndim == 4			)
   self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size			)
   self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length			)
   self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels			)
 def   A_			(   self						:   Optional[int]    ,  UpperCAmelCase						:   Dict			)    ->			List[Any]:
   lowerCamelCase__					:   List[Any]      				=	load_dataset('hf-internal-testing/librispeech_asr_dummy'    ,  'clean'    ,  split='validation'			)
   # automatic decoding with librispeech
   lowerCamelCase__					:   Tuple      				=	ds.sort('id'			).select(range(__A			)			)[:num_samples]["audio"]

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







 def   A_			(   self						:   Union[str, Any]			)    ->			List[Any]:
   lowerCamelCase__					:   Dict      				=	self._load_datasamples(1			)
   lowerCamelCase__					:   Any      				=	TvltFeatureExtractor()
   lowerCamelCase__					:   List[str]      				=	feature_extractor(__A    ,  return_tensors='pt'			).audio_values

   self.assertEquals(audio_values.shape    ,  (1, 1, 192, 128)			)

   lowerCamelCase__					:   Optional[int]      				=	torch.tensor([[-0.3_0_3_2, -0.2_7_0_8], [-0.4_4_3_4, -0.4_0_0_7]]			)
   self.assertTrue(torch.allclose(audio_values[0, 0, :2, :2]    ,  __A    ,  atol=1e-4			)			)












import contextlib
from multiprocessing import Pool, RLock

from tqdm.auto import tqdm

from ..utils import experimental, logging


_UpperCAmelCase       : Union[str, Any]			     =       logging.get_logger(__name__)

class      lowerCAmelCase							:
 UpperCAmelCase__									=	None


@experimental
def  SCREAMING_SNAKE_CASE   ( _UpperCAmelCase		,			_UpperCAmelCase		,			_UpperCAmelCase		,			_UpperCAmelCase		,			_UpperCAmelCase		,			_UpperCAmelCase		,			_UpperCAmelCase				)    ->  Any:
    if ParallelBackendConfig.backend_name is None:
        return _map_with_multiprocessing_pool(
            _UpperCAmelCase		,			_UpperCAmelCase		,			_UpperCAmelCase		,			_UpperCAmelCase		,			_UpperCAmelCase		,			_UpperCAmelCase		,			_UpperCAmelCase				)

    return _map_with_joblib(_UpperCAmelCase		,			_UpperCAmelCase		,			_UpperCAmelCase		,			_UpperCAmelCase		,			_UpperCAmelCase		,			_UpperCAmelCase		,			_UpperCAmelCase				)


def  SCREAMING_SNAKE_CASE   ( _UpperCAmelCase		,			_UpperCAmelCase		,			_UpperCAmelCase		,			_UpperCAmelCase		,			_UpperCAmelCase		,			_UpperCAmelCase		,			_UpperCAmelCase				)    ->  Any:
    lowerCamelCase__					:   int      				=	num_proc if num_proc <= len(_UpperCAmelCase				) else len(_UpperCAmelCase				)
    lowerCamelCase__					:   Optional[int]      				=	[]  # We organize the splits ourselve (contiguous splits)
    for index in range(_UpperCAmelCase				):
        lowerCamelCase__					:   str      				=	len(_UpperCAmelCase				) // num_proc
        lowerCamelCase__					:   List[str]      				=	len(_UpperCAmelCase				) % num_proc
        lowerCamelCase__					:   Tuple      				=	div * index + min(_UpperCAmelCase		,			_UpperCAmelCase				)
        lowerCamelCase__					:   int      				=	start + div + (1 if index < mod else 0)
        split_kwds.append((function, iterable[start:end], types, index, disable_tqdm, desc)				)

    if len(_UpperCAmelCase				) != sum(len(i[1]				) for i in split_kwds				):
        raise ValueError(
            F"""Error dividing inputs iterable among processes. """
            F"""Total number of objects {len(_UpperCAmelCase				)}, """
            F"""length: {sum(len(i[1]				) for i in split_kwds				)}"""				)

    logger.info(
        F"""Spawning {num_proc} processes for {len(_UpperCAmelCase				)} objects in slices of {[len(i[1]				) for i in split_kwds]}"""				)
    lowerCamelCase__				,							lowerCamelCase__					:   List[str]      				=	None, None
    if not disable_tqdm:
        lowerCamelCase__				,							lowerCamelCase__					:   Union[str, Any]      				=	(RLock(),), tqdm.set_lock
    with Pool(_UpperCAmelCase		,			initargs=_UpperCAmelCase		,			initializer=_UpperCAmelCase				) as pool:
        lowerCamelCase__					:   Any      				=	pool.map(_UpperCAmelCase		,			_UpperCAmelCase				)
    logger.info(F"""Finished {num_proc} processes"""				)
    lowerCamelCase__					:   Union[str, Any]      				=	[obj for proc_res in mapped for obj in proc_res]
    logger.info(F"""Unpacked {len(_UpperCAmelCase				)} objects"""				)

    return mapped


def  SCREAMING_SNAKE_CASE   ( _UpperCAmelCase		,			_UpperCAmelCase		,			_UpperCAmelCase		,			_UpperCAmelCase		,			_UpperCAmelCase		,			_UpperCAmelCase		,			_UpperCAmelCase				)    ->  int:
    # progress bar is not yet supported for _map_with_joblib, because tqdm couldn't accurately be applied to joblib,
    # and it requires monkey-patching joblib internal classes which is subject to change
    import joblib

    with joblib.parallel_backend(ParallelBackendConfig.backend_name		,			n_jobs=_UpperCAmelCase				):
        return joblib.Parallel()(
            joblib.delayed(_UpperCAmelCase				)((function, obj, types, None, True, None)				) for obj in iterable				)


@experimental
@contextlib.contextmanager
def  SCREAMING_SNAKE_CASE   ( _UpperCAmelCase				)    ->  Tuple:
    lowerCamelCase__					:   List[str]      				=	backend_name

    if backend_name == "spark":
        from joblibspark import register_spark

        register_spark()

        # TODO: call create_cache_and_write_probe if "download" in steps
        # TODO: raise NotImplementedError when Dataset.map etc is called

    try:
        yield
    finally:
        lowerCamelCase__					:   str      				=	None





