






'''simple docstring'''


import itertools
import random
import unittest

import numpy as np

from transformers import EncodecFeatureExtractor
from transformers.testing_utils import require_torch
from transformers.utils.import_utils import is_torch_available

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


if is_torch_available():
  import torch


SCREAMING_SNAKE_CASE_				      =							random.Random()






def   lowerCAmelCase__			(					SCREAMING_SNAKE_CASE__       ,		SCREAMING_SNAKE_CASE__=1.0       ,		SCREAMING_SNAKE_CASE__=None       ,		SCREAMING_SNAKE_CASE__=None		):
 if rng is None:
  __a    :	Optional[Any]									=						global_rng

 __a    :	Optional[int]									=						[]
 for batch_idx in range(shape[0]		):
  values.append([]		)
  for _ in range(shape[1]		):
   values[-1].append(rng.random() * scale		)

 return values




@require_torch
class 							lowerCAmelCase					(							unittest.TestCase   ):







   """simple docstring"""







   def __init__( self   ,   _A   ,   _A=7   ,   _A=400   ,   _A=2000   ,   _A=1   ,   _A=0.0   ,   _A=24000   ,   _A=True   ,   )    ->   List[Any]:
    __a    :	List[str]									=						parent
    __a    :	str									=						batch_size
    __a    :	List[Any]									=						min_seq_length
    __a    :	List[Any]									=						max_seq_length
    __a    :	Optional[Any]									=						(self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
    __a    :	List[Any]									=						feature_size
    __a    :	List[str]									=						padding_value
    __a    :	Optional[Any]									=						sampling_rate
    __a    :	Tuple									=						return_attention_mask



   def    __magic_name__    ( self			)    ->   Any:
    return {
        "feature_size": self.feature_size,
        "padding_value": self.padding_value,
        "sampling_rate": self.sampling_rate,
        "return_attention_mask": self.return_attention_mask,
    }



   def    __magic_name__    ( self   ,   _A=False   ,   _A=False			)    ->   Dict:
    def _flatten(_A			):
     return list(itertools.chain(*_A			)			)

    if equal_length:
     __a    :	Optional[Any]									=						floats_list((self.batch_size, self.max_seq_length)			)
    else:
     # make sure that inputs increase in size
     __a    :	Optional[int]									=						[
         _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    :	Optional[int]									=						[np.asarray(_A			) for x in audio_inputs]

    return audio_inputs




@require_torch
class 							lowerCAmelCase					(							lowerCamelCase__							,      unittest.TestCase   ):







   """simple docstring"""







   _A						=   EncodecFeatureExtractor



   def    __magic_name__    ( self			)    ->   int:
    __a    :	Union[str, Any]									=						EnCodecFeatureExtractionTester(self			)



   def    __magic_name__    ( self			)    ->   int:
    # Tests that all call wrap to encode_plus and batch_encode_plus
    __a    :	Dict									=						self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()			)
    # create three inputs of length 800, 1000, and 1200
    __a    :	Union[str, Any]									=						[floats_list((1, x)			)[0] for x in range(800   ,   1400   ,   200			)]
    __a    :	Optional[int]									=						[np.asarray(_A			) for audio_input in audio_inputs]

    # Test not batched input
    __a    :	str									=						feat_extract(audio_inputs[0]   ,   return_tensors='np'			).input_values
    __a    :	List[str]									=						feat_extract(np_audio_inputs[0]   ,   return_tensors='np'			).input_values
    self.assertTrue(np.allclose(_A   ,   _A   ,   atol=1E-3			)			)

    # Test batched
    __a    :	Optional[int]									=						feat_extract(_A   ,   padding=_A   ,   return_tensors='np'			).input_values
    __a    :	str									=						feat_extract(_A   ,   padding=_A   ,   return_tensors='np'			).input_values
    for enc_seq_a, enc_seq_a in zip(_A   ,   _A			):
     self.assertTrue(np.allclose(_A   ,   _A   ,   atol=1E-3			)			)



   def    __magic_name__    ( self			)    ->   Any:
    __a    :	int									=						self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()			)
    __a    :	Dict									=						np.random.rand(100			).astype(np.floataa			)
    __a    :	str									=						np_audio_inputs.tolist()

    for inputs in [py_audio_inputs, np_audio_inputs]:
     __a    :	Union[str, Any]									=						feature_extractor.pad([{'input_values': inputs}]   ,   return_tensors='np'			)
     self.assertTrue(np_processed.input_values.dtype == np.floataa			)
     __a    :	int									=						feature_extractor.pad([{'input_values': inputs}]   ,   return_tensors='pt'			)
     self.assertTrue(pt_processed.input_values.dtype == torch.floataa			)



   def    __magic_name__    ( self   ,   _A			)    ->   List[Any]:
    from datasets import load_dataset

    __a    :	Union[str, Any]									=						load_dataset('hf-internal-testing/librispeech_asr_dummy'   ,   'clean'   ,   split='validation'			)
    # automatic decoding with librispeech
    __a    :	Dict									=						ds.sort('id'			).select(range(_A			)			)[:num_samples]['''audio''']

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



   def    __magic_name__    ( self			)    ->   str:
    # fmt: off
    __a    :	Optional[int]									=						torch.tensor(
        [2.3_8_0_4E-0_3, 2.0_7_5_2E-0_3, 1.9_8_3_6E-0_3, 2.1_0_5_7E-0_3, 1.6_1_7_4E-0_3,
         3.0_5_1_8E-0_4, 9.1_5_5_3E-0_5, 3.3_5_6_9E-0_4, 9.7_6_5_6E-0_4, 1.8_3_1_1E-0_3,
         2.0_1_4_2E-0_3, 2.1_0_5_7E-0_3, 1.7_3_9_5E-0_3, 4.5_7_7_6E-0_4, -3.9_6_7_3E-0_4,
         4.5_7_7_6E-0_4, 1.0_0_7_1E-0_3, 9.1_5_5_3E-0_5, 4.8_8_2_8E-0_4, 1.1_5_9_7E-0_3,
         7.3_2_4_2E-0_4, 9.4_6_0_4E-0_4, 1.8_0_0_5E-0_3, 1.8_3_1_1E-0_3, 8.8_5_0_1E-0_4,
         4.2_7_2_5E-0_4, 4.8_8_2_8E-0_4, 7.3_2_4_2E-0_4, 1.0_9_8_6E-0_3, 2.1_0_5_7E-0_3]			)
    # fmt: on
    __a    :	Optional[int]									=						self._load_datasamples(1			)
    __a    :	Optional[Any]									=						EncodecFeatureExtractor()
    __a    :	List[str]									=						feature_extractor(_A   ,   return_tensors='pt'			).input_values
    self.assertEquals(input_values.shape   ,   (1, 1, 93680)			)
    self.assertTrue(torch.allclose(input_values[0, 0, :30]   ,   _A   ,   atol=1E-6			)			)



   def    __magic_name__    ( self			)    ->   List[Any]:
    # fmt: off
    __a    :	Union[str, Any]									=						torch.tensor(
        [2.3_8_0_4E-0_3, 2.0_7_5_2E-0_3, 1.9_8_3_6E-0_3, 2.1_0_5_7E-0_3, 1.6_1_7_4E-0_3,
         3.0_5_1_8E-0_4, 9.1_5_5_3E-0_5, 3.3_5_6_9E-0_4, 9.7_6_5_6E-0_4, 1.8_3_1_1E-0_3,
         2.0_1_4_2E-0_3, 2.1_0_5_7E-0_3, 1.7_3_9_5E-0_3, 4.5_7_7_6E-0_4, -3.9_6_7_3E-0_4,
         4.5_7_7_6E-0_4, 1.0_0_7_1E-0_3, 9.1_5_5_3E-0_5, 4.8_8_2_8E-0_4, 1.1_5_9_7E-0_3,
         7.3_2_4_2E-0_4, 9.4_6_0_4E-0_4, 1.8_0_0_5E-0_3, 1.8_3_1_1E-0_3, 8.8_5_0_1E-0_4,
         4.2_7_2_5E-0_4, 4.8_8_2_8E-0_4, 7.3_2_4_2E-0_4, 1.0_9_8_6E-0_3, 2.1_0_5_7E-0_3]			)
    # fmt: on
    __a    :	List[str]									=						self._load_datasamples(1			)
    __a    :	Optional[Any]									=						[np.tile(input_audio[0][None]   ,   reps=(2, 1)			)]
    input_audio[0][1] *= 0.5
    __a    :	Dict									=						EncodecFeatureExtractor(feature_size=2			)
    __a    :	str									=						feature_extractor(_A   ,   return_tensors='pt'			).input_values
    self.assertEquals(input_values.shape   ,   (1, 2, 93680)			)
    self.assertTrue(torch.allclose(input_values[0, 0, :30]   ,   _A   ,   atol=1E-6			)			)
    self.assertTrue(torch.allclose(input_values[0, 1, :30]   ,   EXPECTED_INPUT_VALUES * 0.5   ,   atol=1E-6			)			)



   def    __magic_name__    ( self			)    ->   int:
    __a    :	Optional[Any]									=						self._load_datasamples(2			)
    # would be easier if the stride was like
    __a    :	Union[str, Any]									=						EncodecFeatureExtractor(feature_size=1   ,   chunk_length_s=1   ,   overlap=0.01			)

    # pad and trunc raise an error ?
    with self.assertRaisesRegex(
        _A   ,   '^Both padding and truncation were set. Make sure you only set one.$'   ,   ):
     __a    :	Union[str, Any]									=						feature_extractor(
         _A   ,   padding='max_length'   ,   truncation=_A   ,   return_tensors='pt'			).input_values

    # truncate to chunk
    __a    :	List[str]									=						feature_extractor(_A   ,   truncation=_A   ,   return_tensors='pt'			).input_values
    self.assertEquals(truncated_outputs.shape   ,   (2, 1, 71520)			)  # 2 chunks

    # force truncate to max_length
    __a    :	int									=						feature_extractor(
        _A   ,   truncation=_A   ,   max_length=48000   ,   return_tensors='pt'			).input_values
    self.assertEquals(truncated_outputs.shape   ,   (2, 1, 48000)			)

    # pad to chunk
    __a    :	List[str]									=						feature_extractor(_A   ,   padding=_A   ,   return_tensors='pt'			).input_values
    self.assertEquals(padded_outputs.shape   ,   (2, 1, 95280)			)

    # pad to chunk
    __a    :	Optional[int]									=						feature_extractor(_A   ,   return_tensors='pt'			).input_values
    self.assertEquals(truncated_outputs.shape   ,   (2, 1, 95280)			)

    # force pad to max length
    __a    :	List[Any]									=						feature_extractor(
        _A   ,   padding='max_length'   ,   max_length=100000   ,   return_tensors='pt'			).input_values
    self.assertEquals(truncated_outputs.shape   ,   (2, 1, 100000)			)

    # force no pad
    with self.assertRaisesRegex(
        _A   ,   '^Unable to create tensor, you should probably activate padding with \'padding=True\' to have batched tensors with the same length.$'   ,   ):
     __a    :	List[Any]									=						feature_extractor(_A   ,   padding=_A   ,   return_tensors='pt'			).input_values

    __a    :	List[Any]									=						feature_extractor(input_audio[0]   ,   padding=_A   ,   return_tensors='pt'			).input_values
    self.assertEquals(truncated_outputs.shape   ,   (1, 1, 93680)			)

    # no pad if no chunk_length_s
    __a    :	List[str]									=						None
    with self.assertRaisesRegex(
        _A   ,   '^Unable to create tensor, you should probably activate padding with \'padding=True\' to have batched tensors with the same length.$'   ,   ):
     __a    :	Union[str, Any]									=						feature_extractor(_A   ,   padding=_A   ,   return_tensors='pt'			).input_values

    __a    :	List[Any]									=						feature_extractor(input_audio[0]   ,   padding=_A   ,   return_tensors='pt'			).input_values
    self.assertEquals(truncated_outputs.shape   ,   (1, 1, 93680)			)

    # no pad if no overlap
    __a    :	str									=						2
    __a    :	str									=						None
    with self.assertRaisesRegex(
        _A   ,   '^Unable to create tensor, you should probably activate padding with \'padding=True\' to have batched tensors with the same length.$'   ,   ):
     __a    :	str									=						feature_extractor(_A   ,   padding=_A   ,   return_tensors='pt'			).input_values

    __a    :	Dict									=						feature_extractor(input_audio[0]   ,   padding=_A   ,   return_tensors='pt'			).input_values
    self.assertEquals(truncated_outputs.shape   ,   (1, 1, 93680)			)













'''simple docstring'''


from typing import List, Optional, Tuple, Union

from . import logging
from .import_utils import is_torch_available, is_torch_version


if is_torch_available():
  import torch

SCREAMING_SNAKE_CASE_				      =							logging.get_logger(__name__)  # pylint: disable=invalid-name

try:
  from torch._dynamo import allow_in_graph as maybe_allow_in_graph
except (ImportError, ModuleNotFoundError):






  def   lowerCAmelCase__			(					cls		):
   return cls






def   lowerCAmelCase__			(					SCREAMING_SNAKE_CASE__       ,		SCREAMING_SNAKE_CASE__ = None       ,		SCREAMING_SNAKE_CASE__ = None       ,		SCREAMING_SNAKE_CASE__ = None       ,		SCREAMING_SNAKE_CASE__ = None       ,		):
 __a    :	int									=						device
 __a    :	Union[str, Any]									=						shape[0]

 __a    :	Union[str, Any]									=						layout or torch.strided
 __a    :	Any									=						device or torch.device('cpu'		)

 if generator is not None:
  __a    :	Optional[int]									=						generator.device.type if not isinstance(SCREAMING_SNAKE_CASE__       ,		SCREAMING_SNAKE_CASE__		) else generator[0].device.type
  if gen_device_type != device.type and gen_device_type == "cpu":
   __a    :	int									=						'cpu'
   if device != "mps":
    logger.info(
        f'''The passed generator was created on \'cpu\' even though a tensor on {device} was expected.'''
        f''' Tensors will be created on \'cpu\' and then moved to {device}. Note that one can probably'''
        f''' slighly speed up this function by passing a generator that was created on the {device} device.'''		)
  elif gen_device_type != device.type and gen_device_type == "cuda":
   raise ValueError(f'''Cannot generate a {device} tensor from a generator of type {gen_device_type}.'''		)

 if isinstance(SCREAMING_SNAKE_CASE__       ,		SCREAMING_SNAKE_CASE__		):
  __a    :	Optional[int]									=						(1,) + shape[1:]
  __a    :	Dict									=						[
      torch.randn(SCREAMING_SNAKE_CASE__       ,		generator=generator[i]       ,		device=SCREAMING_SNAKE_CASE__       ,		dtype=SCREAMING_SNAKE_CASE__       ,		layout=SCREAMING_SNAKE_CASE__		)
      for i in range(SCREAMING_SNAKE_CASE__		)
  ]
  __a    :	List[str]									=						torch.cat(SCREAMING_SNAKE_CASE__       ,		dim=0		).to(SCREAMING_SNAKE_CASE__		)
 else:
  __a    :	Tuple									=						torch.randn(SCREAMING_SNAKE_CASE__       ,		generator=SCREAMING_SNAKE_CASE__       ,		device=SCREAMING_SNAKE_CASE__       ,		dtype=SCREAMING_SNAKE_CASE__       ,		layout=SCREAMING_SNAKE_CASE__		).to(SCREAMING_SNAKE_CASE__		)

 return latents






def   lowerCAmelCase__			(					SCREAMING_SNAKE_CASE__		):
 if is_torch_version('<'       ,		'2.0.0'		) or not hasattr(SCREAMING_SNAKE_CASE__       ,		'_dynamo'		):
  return False
 return isinstance(SCREAMING_SNAKE_CASE__       ,		torch._dynamo.eval_frame.OptimizedModule		)




