




'''simple docstring'''







import timeit
from typing import Callable, Optional

from ..configuration_utils import PretrainedConfig
from ..models.auto.modeling_auto import MODEL_MAPPING, MODEL_WITH_LM_HEAD_MAPPING
from ..utils import is_pyanvml_available, is_torch_available, logging
from .benchmark_utils import (
    Benchmark,
    Memory,
    MemorySummary,
    measure_peak_memory_cpu,
    start_memory_tracing,
    stop_memory_tracing,
)


if is_torch_available():
    import torch

    from .benchmark_args import PyTorchBenchmarkArguments


if is_pyanvml_available():
    import pyanvml.pyanvml as nvml


lowerCAmelCase_    	=							logging.get_logger(__name__)

class   _snake_case( UpperCAmelCase			):
 __snake_case:       PyTorchBenchmarkArguments
 __snake_case:       PretrainedConfig
 __snake_case:       str									=				"PyTorch"






 @property
 def 	_UpperCamelCase   (self					:     Optional[Any]						)					->      List[str]:






    """simple docstring"""


    return torch.__version__






 def 	_UpperCamelCase   (self					:     Optional[Any]						,		a					:     str						,		a					:     int						,		a					:     int						)					->      float:






    """simple docstring"""


    A__         =    self._prepare_inference_func(a						,		a						,		a						)
    return self._measure_speed(_inference						)






 def 	_UpperCamelCase   (self					:     Any						,		a					:     str						,		a					:     int						,		a					:     int						)					->      [Memory, Optional[MemorySummary]]:






    """simple docstring"""


    A__         =    self._prepare_inference_func(a						,		a						,		a						)
    return self._measure_memory(_inference						)






 def 	_UpperCamelCase   (self					:     Optional[Any]						,		a					:     str						,		a					:     int						,		a					:     int						)					->      float:






    """simple docstring"""


    A__         =    self._prepare_train_func(a						,		a						,		a						)
    return self._measure_speed(_train						)






 def 	_UpperCamelCase   (self					:     Tuple						,		a					:     str						,		a					:     int						,		a					:     int						)					->      [Memory, Optional[MemorySummary]]:






    """simple docstring"""


    A__         =    self._prepare_train_func(a						,		a						,		a						)
    return self._measure_memory(_train						)






 def 	_UpperCamelCase   (self					:     str						,		a					:     str						,		a					:     int						,		a					:     int						)					->      Callable[[], None]:






    """simple docstring"""


    A__         =    self.config_dict[model_name]

    if self.args.torchscript:
       A__         =    True

    A__         =    (
        hasattr(a						,		'architectures'						)
        and isinstance(config.architectures						,		a						)
        and len(config.architectures						) > 0
    )
    if not self.args.only_pretrain_model and has_model_class_in_config:
       try:
          A__         =    config.architectures[0]
          A__         =    __import__('transformers'						,		fromlist=[model_class]						)
          A__         =    getattr(a						,		a						)
          A__         =    model_cls(a						)
       except ImportError:
          raise ImportError(
              f"""{model_class} does not exist. If you just want to test the pretrained model, you might want to"""
              ' set `--only_pretrain_model` or `args.only_pretrain_model=True`.'						)
    else:
       A__         =    MODEL_MAPPING[config.__class__](a						)

    model.eval()
    model.to(self.args.device						)

    # encoder-decoder has vocab size saved differently
    A__         =    config.vocab_size if hasattr(a						,		'vocab_size'						) else config.encoder.vocab_size
    A__         =    torch.randint(a						,		(batch_size, sequence_length)						,		dtype=torch.long						,		device=self.args.device						)

    if self.args.fpaa:
       logger.info('Running training in Mixed Precision...'						)
       if not self.args.is_gpu:
          raise ValueError('Mixed precision is possible only for GPU.'						)
       # amp seems to have memory leaks so that memory usage
       # is measured using .half() for now https://github.com/NVIDIA/apex/issues/439
       model.half()

    if self.args.torchscript:
       with torch.no_grad():
          A__         =    torch.jit.trace(a						,		a						)
    else:
       A__         =    model

    def encoder_decoder_forward():
       with torch.no_grad():
          A__         =    inference_model(a						,		decoder_input_ids=a						)
       return outputs

    def encoder_forward():
       with torch.no_grad():
          A__         =    inference_model(a						)
       return outputs

    A__         =    encoder_decoder_forward if config.is_encoder_decoder else encoder_forward
    return _forward






 def 	_UpperCamelCase   (self					:     Optional[Any]						,		a					:     str						,		a					:     int						,		a					:     int						)					->      Callable[[], None]:






    """simple docstring"""


    A__         =    self.config_dict[model_name]

    A__         =    (
        hasattr(a						,		'architectures'						)
        and isinstance(config.architectures						,		a						)
        and len(config.architectures						) > 0
    )
    if not self.args.only_pretrain_model and has_model_class_in_config:
       try:
          A__         =    config.architectures[0]
          A__         =    __import__('transformers'						,		fromlist=[model_class]						)
          A__         =    getattr(a						,		a						)
          A__         =    model_cls(a						)
       except ImportError:
          raise ImportError(
              f"""{model_class} does not exist. If you just want to test the pretrained model, you might want to"""
              ' set `--only_pretrain_model` or `args.only_pretrain_model=True`.'						)
    else:
       A__         =    MODEL_WITH_LM_HEAD_MAPPING[config.__class__](a						)

    if self.args.torchscript:
       raise NotImplementedError('Training for torchscript is currently not implemented'						)
    else:
       A__         =    model

    model.train()
    model.to(self.args.device						)

    # encoder-decoder has vocab size saved differently
    A__         =    config.vocab_size if hasattr(a						,		'vocab_size'						) else config.encoder.vocab_size
    A__         =    torch.randint(a						,		(batch_size, sequence_length)						,		dtype=torch.long						,		device=self.args.device						)

    if self.args.fpaa:
       logger.info('Running training in Mixed Precision...'						)
       if not self.args.is_gpu:
          raise ValueError('Mixed precision is possible only for GPU.'						)

       # amp seems to have memory leaks so that memory usage
       # is measured using .half() for now https://github.com/NVIDIA/apex/issues/439
       model.half()

    def compute_loss_and_backprob_encoder():
       A__         =    train_model(a						,		labels=a						)[0]
       loss.backward()
       return loss

    def compute_loss_and_backprob_encoder_decoder():
       A__         =    train_model(a						,		decoder_input_ids=a						,		labels=a						)[0]
       loss.backward()
       return loss

    A__         =    (
        compute_loss_and_backprob_encoder_decoder
        if config.is_encoder_decoder
        else compute_loss_and_backprob_encoder
    )
    return _train






 def 	_UpperCamelCase   (self					:     Tuple						,		a					:     str						)					->      float:






    """simple docstring"""


    try:
       if self.args.is_tpu or self.args.torchscript:
          # run additional 10 times to stabilize compilation for tpu and torchscript
          logger.info('Do inference on TPU or torchscript. Running model 5 times to stabilize compilation'						)
          timeit.repeat(
              a						,		repeat=1						,		number=5						,		)

       # as written in https://docs.python.org/2/library/timeit.html#timeit.Timer.repeat, min should be taken rather than the average
       A__         =    timeit.repeat(
           a						,		repeat=self.args.repeat						,		number=10						,		)

       if self.args.is_tpu and self.args.torch_xla_tpu_print_metrics:
          import torch_xla.debug.metrics as met

          self.print_fn(met.metrics_report()						)

       return min(a						) / 10.0
    except RuntimeError as e:
       self.print_fn(f"""Doesn't fit on GPU. {e}"""						)
       return "N/A"






 def 	_UpperCamelCase   (self					:     Any						,		a					:     Callable[[], None]						)					->      [Memory, MemorySummary]:






    """simple docstring"""


    try:
       if self.args.trace_memory_line_by_line:
          A__         =    start_memory_tracing('transformers'						)

       if self.args.is_tpu:
          # tpu
          raise NotImplementedError(
              'Memory Benchmarking is currently not implemented for TPU. Please disable memory benchmarking with'
              ' `--no-memory` or `args.memory=False`'						)
       elif self.args.is_gpu:
          if not is_pyanvml_available():
             logger.warning(
                 'py3nvml not installed, we won\'t log GPU memory usage. '
                 'Install py3nvml (pip install py3nvml) to log information about GPU.'						)
             A__         =    'N/A'
          else:
             logger.info(
                 'Measuring total GPU usage on GPU device. Make sure to not have additional processes running'
                 ' on the same GPU.'						)
             # init nvml
             nvml.nvmlInit()
             func()
             A__         =    nvml.nvmlDeviceGetHandleByIndex(self.args.device_idx						)
             A__         =    nvml.nvmlDeviceGetMemoryInfo(a						)
             A__         =    meminfo.used
             A__         =    Memory(a						)
             # shutdown nvml
             nvml.nvmlShutdown()
       else:
          # cpu
          A__         =    measure_peak_memory_cpu(a						)
          A__         =    Memory(a						) if isinstance(a						,		a						) else memory_bytes

       if self.args.trace_memory_line_by_line:
          A__         =    stop_memory_tracing(a						)
       else:
          A__         =    None

       return memory, summary
    except RuntimeError as e:
       self.print_fn(f"""Doesn't fit on GPU. {e}"""						)
       return "N/A", None








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


lowerCAmelCase_    	=							random.Random()







def 	_A			(   UpperCAmelCase			,UpperCAmelCase=1.0			,UpperCAmelCase=None			,UpperCAmelCase=None	):






   '''simple docstring'''







   if rng is None:
      A__         =    global_rng

   A__         =    []
   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   _snake_case( unittest.TestCase			):






 def __init__(self					:     int						,		a					:     str						,		a					:     Optional[int]=7						,		a					:     Optional[Any]=4_00						,		a					:     Any=20_00						,		a					:     Optional[Any]=1						,		a					:     Optional[int]=0.0						,		a					:     Dict=2_40_00						,		a					:     Tuple=True						,		)					->      Optional[Any]:






    """simple docstring"""


    A__         =    parent
    A__         =    batch_size
    A__         =    min_seq_length
    A__         =    max_seq_length
    A__         =    (self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
    A__         =    feature_size
    A__         =    padding_value
    A__         =    sampling_rate
    A__         =    return_attention_mask






 def 	_UpperCamelCase   (self					:     int						)					->      List[str]:






    """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,
    }






 def 	_UpperCamelCase   (self					:     List[Any]						,		a					:     Any=False						,		a					:     Optional[Any]=False						)					->      Tuple:






    """simple docstring"""


    def _flatten(a					:     Optional[Any]						):
       return list(itertools.chain(*a						)						)

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

    return audio_inputs




@require_torch
class   _snake_case( UpperCAmelCase					,   unittest.TestCase			):
 __snake_case:       Union[str, Any]									=				EncodecFeatureExtractor






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






    """simple docstring"""


    A__         =    EnCodecFeatureExtractionTester(self						)






 def 	_UpperCamelCase   (self					:     Dict						)					->      Optional[Any]:






    """simple docstring"""


    A__         =    self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()						)
    # create three inputs of length 800, 1000, and 1200
    A__         =    [floats_list((1, x)						)[0] for x in range(8_00						,		14_00						,		2_00						)]
    A__         =    [np.asarray(a						) for audio_input in audio_inputs]

    # Test not batched input
    A__         =    feat_extract(audio_inputs[0]						,		return_tensors='np'						).input_values
    A__         =    feat_extract(np_audio_inputs[0]						,		return_tensors='np'						).input_values
    self.assertTrue(np.allclose(a						,		a						,		atol=1e-3						)						)

    # Test batched
    A__         =    feat_extract(a						,		padding=a						,		return_tensors='np'						).input_values
    A__         =    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 	_UpperCamelCase   (self					:     Tuple						)					->      Any:






    """simple docstring"""


    A__         =    self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()						)
    A__         =    np.random.rand(1_00						).astype(np.floataa						)
    A__         =    np_audio_inputs.tolist()

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






 def 	_UpperCamelCase   (self					:     Optional[int]						,		a					:     Optional[Any]						)					->      Union[str, Any]:






    """simple docstring"""


    from datasets import load_dataset

    A__         =    load_dataset('hf-internal-testing/librispeech_asr_dummy'						,		'clean'						,		split='validation'						)
    # automatic decoding with librispeech
    A__         =    ds.sort('id'						).select(range(a						)						)[:num_samples]['audio']

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






 def 	_UpperCamelCase   (self					:     Tuple						)					->      Tuple:






    """simple docstring"""


    A__         =    torch.tensor(
        [2.3804e-03, 2.0752e-03, 1.9836e-03, 2.1057e-03, 1.6174e-03,
         3.0518e-04, 9.1553e-05, 3.3569e-04, 9.7656e-04, 1.8311e-03,
         2.0142e-03, 2.1057e-03, 1.7395e-03, 4.5776e-04, -3.9673e-04,
         4.5776e-04, 1.0071e-03, 9.1553e-05, 4.8828e-04, 1.1597e-03,
         7.3242e-04, 9.4604e-04, 1.8005e-03, 1.8311e-03, 8.8501e-04,
         4.2725e-04, 4.8828e-04, 7.3242e-04, 1.0986e-03, 2.1057e-03]						)
    # fmt: on
    A__         =    self._load_datasamples(1						)
    A__         =    EncodecFeatureExtractor()
    A__         =    feature_extractor(a						,		return_tensors='pt'						).input_values
    self.assertEquals(input_values.shape						,		(1, 1, 9_36_80)						)
    self.assertTrue(torch.allclose(input_values[0, 0, :30]						,		a						,		atol=1e-6						)						)






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






    """simple docstring"""


    A__         =    torch.tensor(
        [2.3804e-03, 2.0752e-03, 1.9836e-03, 2.1057e-03, 1.6174e-03,
         3.0518e-04, 9.1553e-05, 3.3569e-04, 9.7656e-04, 1.8311e-03,
         2.0142e-03, 2.1057e-03, 1.7395e-03, 4.5776e-04, -3.9673e-04,
         4.5776e-04, 1.0071e-03, 9.1553e-05, 4.8828e-04, 1.1597e-03,
         7.3242e-04, 9.4604e-04, 1.8005e-03, 1.8311e-03, 8.8501e-04,
         4.2725e-04, 4.8828e-04, 7.3242e-04, 1.0986e-03, 2.1057e-03]						)
    # fmt: on
    A__         =    self._load_datasamples(1						)
    A__         =    [np.tile(input_audio[0][None]						,		reps=(2, 1)						)]
    input_audio[0][1] *= 0.5
    A__         =    EncodecFeatureExtractor(feature_size=2						)
    A__         =    feature_extractor(a						,		return_tensors='pt'						).input_values
    self.assertEquals(input_values.shape						,		(1, 2, 9_36_80)						)
    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 	_UpperCamelCase   (self					:     Union[str, Any]						)					->      Any:






    """simple docstring"""


    A__         =    self._load_datasamples(2						)
    # would be easier if the stride was like
    A__         =    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__         =    feature_extractor(
           a						,		padding='max_length'						,		truncation=a						,		return_tensors='pt'						).input_values

    # truncate to chunk
    A__         =    feature_extractor(a						,		truncation=a						,		return_tensors='pt'						).input_values
    self.assertEquals(truncated_outputs.shape						,		(2, 1, 7_15_20)						)  # 2 chunks

    # force truncate to max_length
    A__         =    feature_extractor(
        a						,		truncation=a						,		max_length=4_80_00						,		return_tensors='pt'						).input_values
    self.assertEquals(truncated_outputs.shape						,		(2, 1, 4_80_00)						)

    # pad to chunk
    A__         =    feature_extractor(a						,		padding=a						,		return_tensors='pt'						).input_values
    self.assertEquals(padded_outputs.shape						,		(2, 1, 9_52_80)						)

    # pad to chunk
    A__         =    feature_extractor(a						,		return_tensors='pt'						).input_values
    self.assertEquals(truncated_outputs.shape						,		(2, 1, 9_52_80)						)

    # force pad to max length
    A__         =    feature_extractor(
        a						,		padding='max_length'						,		max_length=10_00_00						,		return_tensors='pt'						).input_values
    self.assertEquals(truncated_outputs.shape						,		(2, 1, 10_00_00)						)

    # 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__         =    feature_extractor(a						,		padding=a						,		return_tensors='pt'						).input_values

    A__         =    feature_extractor(input_audio[0]						,		padding=a						,		return_tensors='pt'						).input_values
    self.assertEquals(truncated_outputs.shape						,		(1, 1, 9_36_80)						)

    # no pad if no chunk_length_s
    A__         =    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__         =    feature_extractor(a						,		padding=a						,		return_tensors='pt'						).input_values

    A__         =    feature_extractor(input_audio[0]						,		padding=a						,		return_tensors='pt'						).input_values
    self.assertEquals(truncated_outputs.shape						,		(1, 1, 9_36_80)						)

    # no pad if no overlap
    A__         =    2
    A__         =    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__         =    feature_extractor(a						,		padding=a						,		return_tensors='pt'						).input_values

    A__         =    feature_extractor(input_audio[0]						,		padding=a						,		return_tensors='pt'						).input_values
    self.assertEquals(truncated_outputs.shape						,		(1, 1, 9_36_80)						)


