


'''simple docstring'''



from typing import List, Optional, Union

import numpy as np

from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import PaddingStrategy, TensorType, logging


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





class   UpperCAmelCase_    (_UpperCAmelCase    ):





    """simple docstring"""






    lowerCamelCase    :   Optional[int]					=     ['input_values', 'padding_mask']







    def __init__(				self     ,	SCREAMING_SNAKE_CASE_ = 1     ,	SCREAMING_SNAKE_CASE_ = 2_40_00     ,	SCREAMING_SNAKE_CASE_ = 0.0     ,	SCREAMING_SNAKE_CASE_ = None     ,	SCREAMING_SNAKE_CASE_ = None     ,	**SCREAMING_SNAKE_CASE_     ,	)							->	str:
         super().__init__(feature_size=SCREAMING_SNAKE_CASE_     ,	sampling_rate=SCREAMING_SNAKE_CASE_     ,	padding_value=SCREAMING_SNAKE_CASE_     ,	**SCREAMING_SNAKE_CASE_      )
         __lowerCamelCase  :      List[str]														=						chunk_length_s
         __lowerCamelCase  :      Union[str, Any]														=						overlap







    @property
    def  lowercase_							(				self      )							->	Optional[int]:
         if self.chunk_length_s is None:
              return None
         else:
              return int(self.chunk_length_s * self.sampling_rate      )







    @property
    def  lowercase_							(				self      )							->	Optional[int]:
         if self.chunk_length_s is None or self.overlap is None:
              return None
         else:
              return max(1     ,	int((1.0 - self.overlap) * self.chunk_length      )      )







    def __call__(				self     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_ = None     ,	SCREAMING_SNAKE_CASE_ = False     ,	SCREAMING_SNAKE_CASE_ = None     ,	SCREAMING_SNAKE_CASE_ = None     ,	SCREAMING_SNAKE_CASE_ = None     ,	)							->	BatchFeature:
         if sampling_rate is not None:
              if sampling_rate != self.sampling_rate:
                   raise ValueError(
                       f'The model corresponding to this feature extractor: {self} was trained using a sampling rate of'
                       f' {self.sampling_rate}. Please make sure that the provided audio input was sampled with'
                       f' {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.'      )

         if padding and truncation:
              raise ValueError('Both padding and truncation were set. Make sure you only set one.'      )
         elif padding is None:
              # by default let's pad the inputs
              __lowerCamelCase  :      Tuple														=						True

         __lowerCamelCase  :      Tuple														=						bool(
             isinstance(SCREAMING_SNAKE_CASE_     ,	(list, tuple)      ) and (isinstance(raw_audio[0]     ,	(np.ndarray, tuple, list)      ))      )

         if is_batched:
              __lowerCamelCase  :      Tuple														=						[np.asarray(SCREAMING_SNAKE_CASE_     ,	dtype=np.floataa      ).T for audio in raw_audio]
         elif not is_batched and not isinstance(SCREAMING_SNAKE_CASE_     ,	np.ndarray      ):
              __lowerCamelCase  :      List[Any]														=						np.asarray(SCREAMING_SNAKE_CASE_     ,	dtype=np.floataa      )
         elif isinstance(SCREAMING_SNAKE_CASE_     ,	np.ndarray      ) and raw_audio.dtype is np.dtype(np.floataa      ):
              __lowerCamelCase  :      List[Any]														=						raw_audio.astype(np.floataa      )

         # always return batch
         if not is_batched:
              __lowerCamelCase  :      str														=						[np.asarray(SCREAMING_SNAKE_CASE_      ).T]

         # verify inputs are valid
         for idx, example in enumerate(SCREAMING_SNAKE_CASE_      ):
              if example.ndim > 2:
                   raise ValueError(f'Expected input shape (channels, length) but got shape {example.shape}'      )
              if self.feature_size == 1 and example.ndim != 1:
                   raise ValueError(f'Expected mono audio but example has {example.shape[-1]} channels'      )
              if self.feature_size == 2 and example.shape[-1] != 2:
                   raise ValueError(f'Expected stereo audio but example has {example.shape[-1]} channels'      )

         __lowerCamelCase  :      Any														=						None
         __lowerCamelCase  :      Dict														=						BatchFeature({'input_values': raw_audio}      )
         if self.chunk_stride is not None and self.chunk_length is not None and max_length is None:
              if truncation:
                   __lowerCamelCase  :      Optional[Any]														=						min(array.shape[0] for array in raw_audio      )
                   __lowerCamelCase  :      List[str]														=						int(np.floor(max_length / self.chunk_stride      )      )
                   __lowerCamelCase  :      Dict														=						(nb_step - 1) * self.chunk_stride + self.chunk_length
              elif padding:
                   __lowerCamelCase  :      List[Any]														=						max(array.shape[0] for array in raw_audio      )
                   __lowerCamelCase  :      int														=						int(np.ceil(max_length / self.chunk_stride      )      )
                   __lowerCamelCase  :      Union[str, Any]														=						(nb_step - 1) * self.chunk_stride + self.chunk_length
                   __lowerCamelCase  :      Union[str, Any]														=						'max_length'
              else:
                   __lowerCamelCase  :      Union[str, Any]														=						input_values

        # normal padding on batch
         if padded_inputs is None:
              __lowerCamelCase  :      Any														=						self.pad(
                  SCREAMING_SNAKE_CASE_     ,	max_length=SCREAMING_SNAKE_CASE_     ,	truncation=SCREAMING_SNAKE_CASE_     ,	padding=SCREAMING_SNAKE_CASE_     ,	return_attention_mask=SCREAMING_SNAKE_CASE_     ,	)
              if padding:
                   __lowerCamelCase  :      List[str]														=						padded_inputs.pop('attention_mask'      )

         __lowerCamelCase  :      str														=						[]
         for example in padded_inputs.pop('input_values'      ):
              if self.feature_size == 1:
                   __lowerCamelCase  :      Any														=						example[..., None]
              input_values.append(example.T      )

         __lowerCamelCase  :      List[str]														=						input_values
         if return_tensors is not None:
              __lowerCamelCase  :      Union[str, Any]														=						padded_inputs.convert_to_tensors(SCREAMING_SNAKE_CASE_      )

         return padded_inputs














"""simple docstring"""
import shutil
import tempfile
import unittest
from unittest.mock import patch

from transformers import (
    DefaultFlowCallback,
    IntervalStrategy,
    PrinterCallback,
    ProgressCallback,
    Trainer,
    TrainerCallback,
    TrainingArguments,
    is_torch_available,
)
from transformers.testing_utils import require_torch


if is_torch_available():
 from transformers.trainer import DEFAULT_CALLBACKS

 from .test_trainer import RegressionDataset, RegressionModelConfig, RegressionPreTrainedModel



class        UpperCamelCase__ (				a_):







    """simple docstring"""







    def __init__(		self :	Union[str, Any] ):





           '''simple docstring'''

           __magic_name__      				=    []







    def 				a__			(		self :	Tuple				,    UpperCamelCase_ :	Dict				,    UpperCamelCase_ :	Union[str, Any]				,    UpperCamelCase_ :	int				,    **UpperCamelCase_ :	int ):





           '''simple docstring'''

           self.events.append('on_init_end' )







    def 				a__			(		self :	Any				,    UpperCamelCase_ :	int				,    UpperCamelCase_ :	str				,    UpperCamelCase_ :	List[str]				,    **UpperCamelCase_ :	str ):





           '''simple docstring'''

           self.events.append('on_train_begin' )







    def 				a__			(		self :	str				,    UpperCamelCase_ :	Tuple				,    UpperCamelCase_ :	List[str]				,    UpperCamelCase_ :	Optional[int]				,    **UpperCamelCase_ :	Dict ):





           '''simple docstring'''

           self.events.append('on_train_end' )







    def 				a__			(		self :	List[str]				,    UpperCamelCase_ :	Tuple				,    UpperCamelCase_ :	int				,    UpperCamelCase_ :	Tuple				,    **UpperCamelCase_ :	Optional[Any] ):





           '''simple docstring'''

           self.events.append('on_epoch_begin' )







    def 				a__			(		self :	Optional[int]				,    UpperCamelCase_ :	List[Any]				,    UpperCamelCase_ :	int				,    UpperCamelCase_ :	Tuple				,    **UpperCamelCase_ :	List[Any] ):





           '''simple docstring'''

           self.events.append('on_epoch_end' )







    def 				a__			(		self :	Union[str, Any]				,    UpperCamelCase_ :	Union[str, Any]				,    UpperCamelCase_ :	Any				,    UpperCamelCase_ :	Union[str, Any]				,    **UpperCamelCase_ :	Optional[Any] ):





           '''simple docstring'''

           self.events.append('on_step_begin' )







    def 				a__			(		self :	List[str]				,    UpperCamelCase_ :	Optional[Any]				,    UpperCamelCase_ :	int				,    UpperCamelCase_ :	Any				,    **UpperCamelCase_ :	List[Any] ):





           '''simple docstring'''

           self.events.append('on_step_end' )







    def 				a__			(		self :	List[Any]				,    UpperCamelCase_ :	List[str]				,    UpperCamelCase_ :	int				,    UpperCamelCase_ :	Tuple				,    **UpperCamelCase_ :	Dict ):





           '''simple docstring'''

           self.events.append('on_evaluate' )







    def 				a__			(		self :	str				,    UpperCamelCase_ :	Any				,    UpperCamelCase_ :	Optional[int]				,    UpperCamelCase_ :	Dict				,    **UpperCamelCase_ :	Any ):





           '''simple docstring'''

           self.events.append('on_predict' )







    def 				a__			(		self :	Any				,    UpperCamelCase_ :	Optional[int]				,    UpperCamelCase_ :	Union[str, Any]				,    UpperCamelCase_ :	str				,    **UpperCamelCase_ :	Optional[int] ):





           '''simple docstring'''

           self.events.append('on_save' )







    def 				a__			(		self :	Optional[Any]				,    UpperCamelCase_ :	int				,    UpperCamelCase_ :	str				,    UpperCamelCase_ :	Union[str, Any]				,    **UpperCamelCase_ :	Optional[int] ):





           '''simple docstring'''

           self.events.append('on_log' )







    def 				a__			(		self :	List[Any]				,    UpperCamelCase_ :	Optional[int]				,    UpperCamelCase_ :	List[Any]				,    UpperCamelCase_ :	str				,    **UpperCamelCase_ :	List[Any] ):





           '''simple docstring'''

           self.events.append('on_prediction_step' )



@require_torch
class        UpperCamelCase__ (				unittest.TestCase):







    """simple docstring"""







    def 				a__			(		self :	Optional[Any] ):





           '''simple docstring'''

           __magic_name__      				=    tempfile.mkdtemp()







    def 				a__			(		self :	List[Any] ):





           '''simple docstring'''

           shutil.rmtree(self.output_dir )







    def 				a__			(		self :	List[str]				,    UpperCamelCase_ :	Optional[int]=0				,    UpperCamelCase_ :	List[Any]=0				,    UpperCamelCase_ :	Optional[int]=6_4				,    UpperCamelCase_ :	Optional[int]=6_4				,    UpperCamelCase_ :	List[Any]=None				,    UpperCamelCase_ :	Optional[Any]=False				,    **UpperCamelCase_ :	str ):





           '''simple docstring'''

           __magic_name__      				=    RegressionDataset(length=UpperCamelCase_ )
           __magic_name__      				=    RegressionDataset(length=UpperCamelCase_ )
           __magic_name__      				=    RegressionModelConfig(a=UpperCamelCase_				,    b=UpperCamelCase_ )
           __magic_name__      				=    RegressionPreTrainedModel(UpperCamelCase_ )

           __magic_name__      				=    TrainingArguments(self.output_dir				,    disable_tqdm=UpperCamelCase_				,    report_to=[]				,    **UpperCamelCase_ )
           return Trainer(
               UpperCamelCase_				,    UpperCamelCase_				,    train_dataset=UpperCamelCase_				,    eval_dataset=UpperCamelCase_				,    callbacks=UpperCamelCase_				,    )







    def 				a__			(		self :	Optional[int]				,    UpperCamelCase_ :	str				,    UpperCamelCase_ :	Dict ):





           '''simple docstring'''

           self.assertEqual(len(UpperCamelCase_ )				,    len(UpperCamelCase_ ) )

           # Order doesn't matter
           __magic_name__      				=    sorted(UpperCamelCase_				,    key=lambda UpperCamelCase_ : cb.__name__ if isinstance(UpperCamelCase_				,    UpperCamelCase_ ) else cb.__class__.__name__ )
           __magic_name__      				=    sorted(UpperCamelCase_				,    key=lambda UpperCamelCase_ : cb.__name__ if isinstance(UpperCamelCase_				,    UpperCamelCase_ ) else cb.__class__.__name__ )

           for cba, cba in zip(UpperCamelCase_				,    UpperCamelCase_ ):
                  if isinstance(UpperCamelCase_				,    UpperCamelCase_ ) and isinstance(UpperCamelCase_				,    UpperCamelCase_ ):
                         self.assertEqual(UpperCamelCase_				,    UpperCamelCase_ )
                  elif isinstance(UpperCamelCase_				,    UpperCamelCase_ ) and not isinstance(UpperCamelCase_				,    UpperCamelCase_ ):
                         self.assertEqual(UpperCamelCase_				,    cba.__class__ )
                  elif not isinstance(UpperCamelCase_				,    UpperCamelCase_ ) and isinstance(UpperCamelCase_				,    UpperCamelCase_ ):
                         self.assertEqual(cba.__class__				,    UpperCamelCase_ )
                  else:
                         self.assertEqual(UpperCamelCase_				,    UpperCamelCase_ )







    def 				a__			(		self :	Union[str, Any]				,    UpperCamelCase_ :	Any ):





           '''simple docstring'''

           __magic_name__      				=    ['on_init_end', 'on_train_begin']
           __magic_name__      				=    0
           __magic_name__      				=    len(trainer.get_eval_dataloader() )
           __magic_name__      				=    ['on_prediction_step'] * len(trainer.get_eval_dataloader() ) + ['on_log', 'on_evaluate']
           for _ in range(trainer.state.num_train_epochs ):
                  expected_events.append('on_epoch_begin' )
                  for _ in range(UpperCamelCase_ ):
                         step += 1
                         expected_events += ["on_step_begin", "on_step_end"]
                         if step % trainer.args.logging_steps == 0:
                                expected_events.append('on_log' )
                         if trainer.args.evaluation_strategy == IntervalStrategy.STEPS and step % trainer.args.eval_steps == 0:
                                expected_events += evaluation_events.copy()
                         if step % trainer.args.save_steps == 0:
                                expected_events.append('on_save' )
                  expected_events.append('on_epoch_end' )
                  if trainer.args.evaluation_strategy == IntervalStrategy.EPOCH:
                         expected_events += evaluation_events.copy()
           expected_events += ["on_log", "on_train_end"]
           return expected_events







    def 				a__			(		self :	List[Any] ):





           '''simple docstring'''

           __magic_name__      				=    self.get_trainer()
           __magic_name__      				=    DEFAULT_CALLBACKS.copy() + [ProgressCallback]
           self.check_callbacks_equality(trainer.callback_handler.callbacks				,    UpperCamelCase_ )

           # Callbacks passed at init are added to the default callbacks
           __magic_name__      				=    self.get_trainer(callbacks=[MyTestTrainerCallback] )
           expected_callbacks.append(UpperCamelCase_ )
           self.check_callbacks_equality(trainer.callback_handler.callbacks				,    UpperCamelCase_ )

           # TrainingArguments.disable_tqdm controls if use ProgressCallback or PrinterCallback
           __magic_name__      				=    self.get_trainer(disable_tqdm=UpperCamelCase_ )
           __magic_name__      				=    DEFAULT_CALLBACKS.copy() + [PrinterCallback]
           self.check_callbacks_equality(trainer.callback_handler.callbacks				,    UpperCamelCase_ )







    def 				a__			(		self :	List[str] ):





           '''simple docstring'''

           __magic_name__      				=    DEFAULT_CALLBACKS.copy() + [ProgressCallback]
           __magic_name__      				=    self.get_trainer()

           # We can add, pop, or remove by class name
           trainer.remove_callback(UpperCamelCase_ )
           expected_callbacks.remove(UpperCamelCase_ )
           self.check_callbacks_equality(trainer.callback_handler.callbacks				,    UpperCamelCase_ )

           __magic_name__      				=    self.get_trainer()
           __magic_name__      				=    trainer.pop_callback(UpperCamelCase_ )
           self.assertEqual(cb.__class__				,    UpperCamelCase_ )
           self.check_callbacks_equality(trainer.callback_handler.callbacks				,    UpperCamelCase_ )

           trainer.add_callback(UpperCamelCase_ )
           expected_callbacks.insert(0				,    UpperCamelCase_ )
           self.check_callbacks_equality(trainer.callback_handler.callbacks				,    UpperCamelCase_ )

           # We can also add, pop, or remove by instance
           __magic_name__      				=    self.get_trainer()
           __magic_name__      				=    trainer.callback_handler.callbacks[0]
           trainer.remove_callback(UpperCamelCase_ )
           expected_callbacks.remove(UpperCamelCase_ )
           self.check_callbacks_equality(trainer.callback_handler.callbacks				,    UpperCamelCase_ )

           __magic_name__      				=    self.get_trainer()
           __magic_name__      				=    trainer.callback_handler.callbacks[0]
           __magic_name__      				=    trainer.pop_callback(UpperCamelCase_ )
           self.assertEqual(UpperCamelCase_				,    UpperCamelCase_ )
           self.check_callbacks_equality(trainer.callback_handler.callbacks				,    UpperCamelCase_ )

           trainer.add_callback(UpperCamelCase_ )
           expected_callbacks.insert(0				,    UpperCamelCase_ )
           self.check_callbacks_equality(trainer.callback_handler.callbacks				,    UpperCamelCase_ )







    def 				a__			(		self :	Any ):





           '''simple docstring'''

           import warnings

           # XXX: for now ignore scatter_gather warnings in this test since it's not relevant to what's being tested
           warnings.simplefilter(action='ignore'				,    category=UpperCamelCase_ )

           __magic_name__      				=    self.get_trainer(callbacks=[MyTestTrainerCallback] )
           trainer.train()
           __magic_name__      				=    trainer.callback_handler.callbacks[-2].events
           self.assertEqual(UpperCamelCase_				,    self.get_expected_events(UpperCamelCase_ ) )

           # Independent log/save/eval
           __magic_name__      				=    self.get_trainer(callbacks=[MyTestTrainerCallback]				,    logging_steps=5 )
           trainer.train()
           __magic_name__      				=    trainer.callback_handler.callbacks[-2].events
           self.assertEqual(UpperCamelCase_				,    self.get_expected_events(UpperCamelCase_ ) )

           __magic_name__      				=    self.get_trainer(callbacks=[MyTestTrainerCallback]				,    save_steps=5 )
           trainer.train()
           __magic_name__      				=    trainer.callback_handler.callbacks[-2].events
           self.assertEqual(UpperCamelCase_				,    self.get_expected_events(UpperCamelCase_ ) )

           __magic_name__      				=    self.get_trainer(callbacks=[MyTestTrainerCallback]				,    eval_steps=5				,    evaluation_strategy='steps' )
           trainer.train()
           __magic_name__      				=    trainer.callback_handler.callbacks[-2].events
           self.assertEqual(UpperCamelCase_				,    self.get_expected_events(UpperCamelCase_ ) )

           __magic_name__      				=    self.get_trainer(callbacks=[MyTestTrainerCallback]				,    evaluation_strategy='epoch' )
           trainer.train()
           __magic_name__      				=    trainer.callback_handler.callbacks[-2].events
           self.assertEqual(UpperCamelCase_				,    self.get_expected_events(UpperCamelCase_ ) )

           # A bit of everything
           __magic_name__      				=    self.get_trainer(
               callbacks=[MyTestTrainerCallback]				,    logging_steps=3				,    save_steps=1_0				,    eval_steps=5				,    evaluation_strategy='steps'				,    )
           trainer.train()
           __magic_name__      				=    trainer.callback_handler.callbacks[-2].events
           self.assertEqual(UpperCamelCase_				,    self.get_expected_events(UpperCamelCase_ ) )

           # warning should be emitted for duplicated callbacks
           with patch('transformers.trainer_callback.logger.warning' ) as warn_mock:
                  __magic_name__      				=    self.get_trainer(
                      callbacks=[MyTestTrainerCallback, MyTestTrainerCallback]				,    )
                  assert str(UpperCamelCase_ ) in warn_mock.call_args[0][0]