


import json
import os
import subprocess
import unittest
from ast import literal_eval

import pytest
from parameterized import parameterized_class

from . import is_sagemaker_available


if is_sagemaker_available():
			from sagemaker import Session, TrainingJobAnalytics
			from sagemaker.huggingface import HuggingFace

@pytest.mark.skipif(
    literal_eval(os.getenv("TEST_SAGEMAKER"					,    "False"				)				) is not True					,    reason="Skipping test because should only be run when releasing minor transformers version"					,    )
@pytest.mark.usefixtures("sm_env"				)
@parameterized_class(
    [
        {
            "framework": "pytorch",
            "script": "run_glue.py",
            "model_name_or_path": "distilbert-base-cased",
            "instance_type": "ml.g4dn.xlarge",
            "results": {"train_runtime": 650, "eval_accuracy": 0.6, "eval_loss": 0.9},
        },
        {
            "framework": "tensorflow",
            "script": "run_tf.py",
            "model_name_or_path": "distilbert-base-cased",
            "instance_type": "ml.g4dn.xlarge",
            "results": {"train_runtime": 600, "eval_accuracy": 0.3, "eval_loss": 0.9},
        },
    ]				)
class     a		(unittest.TestCase				):




							"""simple docstring"""


							def    __snake_case (							self	:  Any    )     ->   Tuple:
														if self.framework == "pytorch":
																					subprocess.run(
																					    F'cp ./examples/pytorch/text-classification/run_glue.py {self.env.test_path}/run_glue.py'.split()   ,     encoding="utf-8"   ,     check=lowerCamelCase   ,     )
														assert hasattr(self   ,     "env"    )

							def    __snake_case (							self	:  Dict   ,     lowerCamelCase	:  Union[str, Any]=1    )     ->   Optional[Any]:
														# creates estimator
														return HuggingFace(
														    entry_point=self.script   ,     source_dir=self.env.test_path   ,     role=self.env.role   ,     image_uri=self.env.image_uri   ,     base_job_name=F'{self.env.base_job_name}-single'   ,     instance_count=lowerCamelCase   ,     instance_type=self.instance_type   ,     debugger_hook_config=lowerCamelCase   ,     hyperparameters={**self.env.hyperparameters, "model_name_or_path": self.model_name_or_path}   ,     metric_definitions=self.env.metric_definitions   ,     py_version="py36"   ,     )

							def    __snake_case (							self	:  int   ,     lowerCamelCase	:  int    )     ->   Tuple:
														TrainingJobAnalytics(lowerCamelCase    ).export_csv(F'{self.env.test_path}/{job_name}_metrics.csv'    )






							def    __snake_case (							self	:  Union[str, Any]    )     ->   List[Any]:
														# create estimator
														__snake_case		:      List[Any]										=       self.create_estimator()

														# run training
														estimator.fit()

														# result dataframe
														__snake_case		:      Dict										=       TrainingJobAnalytics(estimator.latest_training_job.name    ).dataframe()

														# extract kpis
														__snake_case		:      Any										=       list(result_metrics_df[result_metrics_df.metric_name == "eval_accuracy"]["value"]    )
														__snake_case		:      List[Any]										=       list(result_metrics_df[result_metrics_df.metric_name == "eval_loss"]["value"]    )
														# get train time from SageMaker job, this includes starting, preprocessing, stopping
														__snake_case		:      Dict										=       (
														    Session().describe_training_job(estimator.latest_training_job.name    ).get("TrainingTimeInSeconds"   ,     999999    )
														)

														# assert kpis
														assert train_runtime <= self.results["train_runtime"]
														assert all(t >= self.results["eval_accuracy"] for t in eval_accuracy    )
														assert all(t <= self.results["eval_loss"] for t in eval_loss    )

														# dump tests result into json file to share in PR
														with open(F'{estimator.latest_training_job.name}.json'   ,     "w"    ) as outfile:
																					json.dump({"train_time": train_runtime, "eval_accuracy": eval_accuracy, "eval_loss": eval_loss}   ,     lowerCamelCase    )





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

import numpy as np
import torch

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


_snake_case					:		List[Any]      =    logging.get_logger(__name__)

class     a		(_lowerCAmelCase				):




							"""simple docstring"""


							__UpperCAmelCase   :							Union[str, Any]					  =  ["input_features", "is_longer"]

							def __init__(							self	:  Optional[int]   ,     lowerCamelCase	:  Any=64   ,     lowerCamelCase	:  Dict=48000   ,     lowerCamelCase	:  Dict=480   ,     lowerCamelCase	:  Tuple=10   ,     lowerCamelCase	:  Optional[int]=1024   ,     lowerCamelCase	:  int=0.0   ,     lowerCamelCase	:  Any=False   ,     lowerCamelCase	:  float = 0   ,     lowerCamelCase	:  float = 14000   ,     lowerCamelCase	:  int = None   ,     lowerCamelCase	:  str = "fusion"   ,     lowerCamelCase	:  str = "repeatpad"   ,     **lowerCamelCase	:  Optional[int]   ,     )     ->   Dict:
														super().__init__(
														    feature_size=lowerCamelCase   ,     sampling_rate=lowerCamelCase   ,     padding_value=lowerCamelCase   ,     return_attention_mask=lowerCamelCase   ,     **lowerCamelCase   ,     )
														__snake_case		:      Optional[Any]										=       top_db
														__snake_case		:      Dict										=       truncation
														__snake_case		:      Dict										=       padding
														__snake_case		:      Optional[Any]										=       fft_window_size
														__snake_case		:      Optional[Any]										=       (fft_window_size >> 1) + 1
														__snake_case		:      Dict										=       hop_length
														__snake_case		:      Optional[int]										=       max_length_s
														__snake_case		:      Optional[int]										=       max_length_s * sampling_rate
														__snake_case		:      Dict										=       sampling_rate
														__snake_case		:      Optional[int]										=       frequency_min
														__snake_case		:      Any										=       frequency_max
														__snake_case		:      Optional[int]										=       mel_filter_bank(
														    num_frequency_bins=self.nb_frequency_bins   ,     num_mel_filters=lowerCamelCase   ,     min_frequency=lowerCamelCase   ,     max_frequency=lowerCamelCase   ,     sampling_rate=lowerCamelCase   ,     norm=lowerCamelCase   ,     mel_scale="htk"   ,     )
														__snake_case		:      Tuple										=       mel_filter_bank(
														    num_frequency_bins=self.nb_frequency_bins   ,     num_mel_filters=lowerCamelCase   ,     min_frequency=lowerCamelCase   ,     max_frequency=lowerCamelCase   ,     sampling_rate=lowerCamelCase   ,     norm="slaney"   ,     mel_scale="slaney"   ,     )

							def    __snake_case (							self	:  str    )     ->   Dict[str, Any]:
														__snake_case		:      List[str]										=       copy.deepcopy(self.__dict__    )
														__snake_case		:      List[Any]										=       self.__class__.__name__
														if "mel_filters" in output:
																					del output["mel_filters"]
														if "mel_filters_slaney" in output:
																					del output["mel_filters_slaney"]
														return output

							def    __snake_case (							self	:  List[Any]   ,     lowerCamelCase	:  np.array   ,     lowerCamelCase	:  Optional[np.array] = None    )     ->   np.ndarray:
														__snake_case		:      List[Any]										=       spectrogram(
														    lowerCamelCase   ,     window_function(self.fft_window_size   ,     "hann"    )   ,     frame_length=self.fft_window_size   ,     hop_length=self.hop_length   ,     power=2.0   ,     mel_filters=lowerCamelCase   ,     log_mel="dB"   ,     )
														return log_mel_spectrogram.T

							def    __snake_case (							self	:  List[Any]   ,     lowerCamelCase	:  Any   ,     lowerCamelCase	:  Union[str, Any]   ,     lowerCamelCase	:  Any    )     ->   str:
														__snake_case		:      Tuple										=       np.array_split(list(range(0   ,     total_frames - chunk_frames + 1    )    )   ,     3    )
														if len(ranges[1]    ) == 0:
																					# if the audio is too short, we just use the first chunk
																					__snake_case		:      Tuple										=       [0]
														if len(ranges[2]    ) == 0:
																					# if the audio is too short, we just use the first chunk
																					__snake_case		:      Tuple										=       [0]
														# randomly choose index for each part
														__snake_case		:      List[Any]										=       np.random.choice(ranges[0]    )
														__snake_case		:      int										=       np.random.choice(ranges[1]    )
														__snake_case		:      List[str]										=       np.random.choice(ranges[2]    )

														__snake_case		:      Dict										=       mel[idx_front : idx_front + chunk_frames, :]
														__snake_case		:      Optional[Any]										=       mel[idx_middle : idx_middle + chunk_frames, :]
														__snake_case		:      Tuple										=       mel[idx_back : idx_back + chunk_frames, :]

														__snake_case		:      Optional[Any]										=       torch.tensor(mel[None, None, :]    )
														__snake_case		:      Optional[int]										=       torch.nn.functional.interpolate(
														    lowerCamelCase   ,     size=[chunk_frames, 64]   ,     mode="bilinear"   ,     align_corners=lowerCamelCase    )
														__snake_case		:      List[Any]										=       mel_shrink[0][0].numpy()
														__snake_case		:      Union[str, Any]										=       np.stack([mel_shrink, mel_chunk_front, mel_chunk_middle, mel_chunk_back]   ,     axis=0    )
														return mel_fusion

							def    __snake_case (							self	:  Any   ,     lowerCamelCase	:  np.array   ,     lowerCamelCase	:  str   ,     lowerCamelCase	:  Tuple   ,     lowerCamelCase	:  Dict    )     ->   np.array:
														if waveform.shape[0] > max_length:
																					if truncation == "rand_trunc":
																												__snake_case		:      List[Any]										=       True
																												# random crop to max_length (for compatibility) -> this should be handled by self.pad
																												__snake_case		:      Tuple										=       len(lowerCamelCase    ) - max_length
																												__snake_case		:      List[Any]										=       np.random.randint(0   ,     overflow + 1    )
																												__snake_case		:      Dict										=       waveform[idx : idx + max_length]
																												__snake_case		:      Optional[int]										=       self._np_extract_fbank_features(lowerCamelCase   ,     self.mel_filters_slaney    )[None, :]
																					elif truncation == "fusion":
																												__snake_case		:      Any										=       self._np_extract_fbank_features(lowerCamelCase   ,     self.mel_filters    )
																												__snake_case		:      Optional[int]										=       max_length // self.hop_length + 1  # the +1 related to how the spectrogram is computed
																												__snake_case		:      str										=       mel.shape[0]
																												if chunk_frames == total_frames:
																																			# there is a corner case where the audio length is larger than max_length but smaller than max_length+hop_length.
																																			# In this case, we just use the whole audio.
																																			__snake_case		:      str										=       np.stack([mel, mel, mel, mel]   ,     axis=0    )
																																			__snake_case		:      Optional[Any]										=       False
																												else:
																																			__snake_case		:      Any										=       self._random_mel_fusion(lowerCamelCase   ,     lowerCamelCase   ,     lowerCamelCase    )
																																			__snake_case		:      Tuple										=       True
																					else:
																												raise NotImplementedError(F'data_truncating {truncation} not implemented'    )

														else:
																					__snake_case		:      Dict										=       False
																					# only use repeat as a new possible value for padding. you repeat the audio before applying the usual max_length padding
																					if waveform.shape[0] < max_length:
																												if padding == "repeat":
																																			__snake_case		:      List[str]										=       int(max_length / len(lowerCamelCase    )    )
																																			__snake_case		:      Any										=       np.stack(np.tile(lowerCamelCase   ,     n_repeat + 1    )    )[:max_length]
																												if padding == "repeatpad":
																																			__snake_case		:      str										=       int(max_length / len(lowerCamelCase    )    )
																																			__snake_case		:      List[str]										=       np.stack(np.tile(lowerCamelCase   ,     lowerCamelCase    )    )
																												__snake_case		:      str										=       np.pad(lowerCamelCase   ,     (0, max_length - waveform.shape[0])   ,     mode="constant"   ,     constant_values=0    )

																					if truncation == "fusion":
																												__snake_case		:      List[str]										=       self._np_extract_fbank_features(lowerCamelCase   ,     self.mel_filters    )
																												__snake_case		:      List[str]										=       np.stack([input_mel, input_mel, input_mel, input_mel]   ,     axis=0    )
																					else:
																												__snake_case		:      Optional[int]										=       self._np_extract_fbank_features(lowerCamelCase   ,     self.mel_filters_slaney    )[None, :]

														return input_mel, longer






							def __call__(							self	:  List[str]   ,     lowerCamelCase	:  Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]   ,     lowerCamelCase	:  str = None   ,     lowerCamelCase	:  Optional[str] = None   ,     lowerCamelCase	:  Optional[int] = None   ,     lowerCamelCase	:  Optional[int] = None   ,     lowerCamelCase	:  Optional[Union[str, TensorType]] = None   ,     **lowerCamelCase	:  Any   ,     )     ->   BatchFeature:
														__snake_case		:      Union[str, Any]										=       truncation if truncation is not None else self.truncation
														__snake_case		:      int										=       padding if padding else self.padding

														if sampling_rate is not None:
																					if sampling_rate != self.sampling_rate:
																												raise ValueError(
																												    F'The model corresponding to this feature extractor: {self.__class__.__name__} was trained using a'
																												    F' sampling rate of {self.sampling_rate}. Please make sure that the provided `raw_speech` input'
																												    F' was sampled with {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."    )

														__snake_case		:      Any										=       isinstance(lowerCamelCase   ,     np.ndarray    ) and len(raw_speech.shape    ) > 1
														if is_batched_numpy and len(raw_speech.shape    ) > 2:
																					raise ValueError(F'Only mono-channel audio is supported for input to {self}'    )
														__snake_case		:      str										=       is_batched_numpy or (
														    isinstance(lowerCamelCase   ,     (list, tuple)    ) and (isinstance(raw_speech[0]   ,     (np.ndarray, tuple, list)    ))
														)

														if is_batched:
																					__snake_case		:      Tuple										=       [np.asarray(lowerCamelCase   ,     dtype=np.floataa    ) for speech in raw_speech]
														elif not is_batched and not isinstance(lowerCamelCase   ,     np.ndarray    ):
																					__snake_case		:      str										=       np.asarray(lowerCamelCase   ,     dtype=np.floataa    )
														elif isinstance(lowerCamelCase   ,     np.ndarray    ) and raw_speech.dtype is np.dtype(np.floataa    ):
																					__snake_case		:      int										=       raw_speech.astype(np.floataa    )

														# always return batch
														if not is_batched:
																					__snake_case		:      Union[str, Any]										=       [np.asarray(lowerCamelCase    )]

														# convert to mel spectrogram, truncate and pad if needed.
														__snake_case		:      Optional[int]										=       [
														    self._get_input_mel(lowerCamelCase   ,     max_length if max_length else self.nb_max_samples   ,     lowerCamelCase   ,     lowerCamelCase    )
														    for waveform in raw_speech
														]

														__snake_case		:      Optional[int]										=       []
														__snake_case		:      Any										=       []
														for mel, longer in padded_inputs:
																					input_mel.append(lowerCamelCase    )
																					is_longer.append(lowerCamelCase    )

														if truncation == "fusion" and sum(lowerCamelCase    ) == 0:
																					# if no audio is longer than 10s, then randomly select one audio to be longer
																					__snake_case		:      Optional[Any]										=       np.random.randint(0   ,     len(lowerCamelCase    )    )
																					__snake_case		:      Union[str, Any]										=       True

														if isinstance(input_mel[0]   ,     lowerCamelCase    ):
																					__snake_case		:      List[str]										=       [np.asarray(lowerCamelCase   ,     dtype=np.floataa    ) for feature in input_mel]

														# is_longer is a list of bool
														__snake_case		:      Any										=       [[longer] for longer in is_longer]

														__snake_case		:      Tuple										=       {"input_features": input_mel, "is_longer": is_longer}
														__snake_case		:      List[str]										=       BatchFeature(lowerCamelCase    )

														if return_tensors is not None:
																					__snake_case		:      Any										=       input_features.convert_to_tensors(lowerCamelCase    )

														return input_features

