






from transformers import BertTokenizer, EncoderDecoderModel, SeqaSeqTrainer, SeqaSeqTrainingArguments
from transformers.testing_utils import TestCasePlus, require_torch, slow
from transformers.utils import is_datasets_available


if is_datasets_available():
  import datasets
class 							A_    (						__lowercase			):






 '''simple docstring'''




 @slow
 @require_torch
 def  snake_case__				(	self) ->		Dict:

     """simple docstring"""



     _UpperCAmelCase				:      Dict			  =				EncoderDecoderModel.from_encoder_decoder_pretrained('''prajjwal1/bert-tiny'''							,  '''prajjwal1/bert-tiny''')
     _UpperCAmelCase				:      str			  =				BertTokenizer.from_pretrained('''bert-base-uncased''')

     _UpperCAmelCase				:      int			  =				bertabert.config.encoder.vocab_size
     _UpperCAmelCase				:      Optional[Any]			  =				tokenizer.sep_token_id
     _UpperCAmelCase				:      Tuple			  =				tokenizer.cls_token_id
     _UpperCAmelCase				:      List[str]			  =				128

     _UpperCAmelCase				:      Tuple			  =				datasets.load_dataset('''cnn_dailymail'''							,  '''3.0.0'''							,  split='''train[:1%]''')
     _UpperCAmelCase				:      Optional[int]			  =				datasets.load_dataset('''cnn_dailymail'''							,  '''3.0.0'''							,  split='''validation[:1%]''')

     _UpperCAmelCase				:      Dict			  =				train_dataset.select(range(32))
     _UpperCAmelCase				:      int			  =				val_dataset.select(range(16))

     _UpperCAmelCase				:      Tuple			  =				4

     def _map_to_encoder_decoder_inputs(_A):
         # Tokenizer will automatically set [BOS] <text> [EOS]
         _UpperCAmelCase				:      Any			  =				tokenizer(batch['''article''']							,  padding='''max_length'''							,  truncation=_A							,  max_length=512)
         _UpperCAmelCase				:      Optional[int]			  =				tokenizer(batch['''highlights''']							,  padding='''max_length'''							,  truncation=_A							,  max_length=128)
         _UpperCAmelCase				:      str			  =				inputs.input_ids
         _UpperCAmelCase				:      Dict			  =				inputs.attention_mask

         _UpperCAmelCase				:      Tuple			  =				outputs.input_ids
         _UpperCAmelCase				:      Tuple			  =				outputs.input_ids.copy()
         _UpperCAmelCase				:      str			  =				[
             [-100 if token == tokenizer.pad_token_id else token for token in labels] for labels in batch['''labels''']
         ]
         _UpperCAmelCase				:      Tuple			  =				outputs.attention_mask

         assert all(len(_A) == 512 for x in inputs.input_ids)
         assert all(len(_A) == 128 for x in outputs.input_ids)

         return batch

     def _compute_metrics(_A):
         _UpperCAmelCase				:      Tuple			  =				pred.label_ids
         _UpperCAmelCase				:      Optional[int]			  =				pred.predictions

         # all unnecessary tokens are removed
         _UpperCAmelCase				:      Any			  =				tokenizer.batch_decode(_A							,  skip_special_tokens=_A)
         _UpperCAmelCase				:      Optional[int]			  =				tokenizer.batch_decode(_A							,  skip_special_tokens=_A)

         _UpperCAmelCase				:      Dict			  =				sum([int(pred_str[i] == label_str[i]) for i in range(len(_A))]) / len(_A)

         return {"accuracy": accuracy}

     # map train dataset
     _UpperCAmelCase				:      Any			  =				train_dataset.map(
         _map_to_encoder_decoder_inputs							,  batched=_A							,  batch_size=_A							,  remove_columns=['''article''', '''highlights''']							,  )
     train_dataset.set_format(
         type='''torch'''							,  columns=['''input_ids''', '''attention_mask''', '''decoder_input_ids''', '''decoder_attention_mask''', '''labels''']							,  )

     # same for validation dataset
     _UpperCAmelCase				:      List[Any]			  =				val_dataset.map(
         _map_to_encoder_decoder_inputs							,  batched=_A							,  batch_size=_A							,  remove_columns=['''article''', '''highlights''']							,  )
     val_dataset.set_format(
         type='''torch'''							,  columns=['''input_ids''', '''attention_mask''', '''decoder_input_ids''', '''decoder_attention_mask''', '''labels''']							,  )

     _UpperCAmelCase				:      Union[str, Any]			  =				self.get_auto_remove_tmp_dir()

     _UpperCAmelCase				:      Union[str, Any]			  =				SeqaSeqTrainingArguments(
         output_dir=_A							,  per_device_train_batch_size=_A							,  per_device_eval_batch_size=_A							,  predict_with_generate=_A							,  evaluation_strategy='''steps'''							,  do_train=_A							,  do_eval=_A							,  warmup_steps=0							,  eval_steps=2							,  logging_steps=2							,  )

     # instantiate trainer
     _UpperCAmelCase				:      str			  =				SeqaSeqTrainer(
         model=_A							,  args=_A							,  compute_metrics=_compute_metrics							,  train_dataset=_A							,  eval_dataset=_A							,  tokenizer=_A							,  )

     # start training
     trainer.train()












from collections import UserDict
from typing import Union

import numpy as np
import requests

from ..utils import (
    add_end_docstrings,
    logging,
)
from .audio_classification import ffmpeg_read
from .base import PIPELINE_INIT_ARGS, Pipeline


SCREAMING_SNAKE_CASE 							=						logging.get_logger(__name__)
@add_end_docstrings(__lowercase			)
class 							A_    (						__lowercase			):






 '''simple docstring'''




 def __init__(	self							,  **_A) ->		int:

     """simple docstring"""



     super().__init__(**_A)

     if self.framework != "pt":
         raise ValueError(f'''The {self.__class__} is only available in PyTorch.''')
     # No specific FOR_XXX available yet


 def __call__(	self							,  _A							,  **_A) ->		List[str]:

     """simple docstring"""



     return super().__call__(_A							,  **_A)


 def  snake_case__				(	self							,  **_A) ->		List[str]:

     """simple docstring"""



     _UpperCAmelCase				:      Optional[int]			  =				{}
     if "candidate_labels" in kwargs:
         _UpperCAmelCase				:      Union[str, Any]			  =				kwargs['''candidate_labels''']
     if "hypothesis_template" in kwargs:
         _UpperCAmelCase				:      Dict			  =				kwargs['''hypothesis_template''']

     return preprocess_params, {}, {}


 def  snake_case__				(	self							,  _A							,  _A=None							,  _A="This is a sound of {}.") ->		Tuple:

     """simple docstring"""



     if isinstance(_A							,  _A):
         if audio.startswith('''http://''') or audio.startswith('''https://'''):
             # We need to actually check for a real protocol, otherwise it's impossible to use a local file
             # like http_huggingface_co.png
             _UpperCAmelCase				:      List[Any]			  =				requests.get(_A).content
         else:
             with open(_A							,  '''rb''') as f:
                 _UpperCAmelCase				:      List[str]			  =				f.read()

     if isinstance(_A							,  _A):
         _UpperCAmelCase				:      List[Any]			  =				ffmpeg_read(_A							,  self.feature_extractor.sampling_rate)

     if not isinstance(_A							,  np.ndarray):
         raise ValueError('''We expect a numpy ndarray as input''')
     if len(audio.shape) != 1:
         raise ValueError('''We expect a single channel audio input for ZeroShotAudioClassificationPipeline''')

     _UpperCAmelCase				:      Optional[Any]			  =				self.feature_extractor(
         [audio]							,  sampling_rate=self.feature_extractor.sampling_rate							,  return_tensors='''pt''')
     _UpperCAmelCase				:      Dict			  =				candidate_labels
     _UpperCAmelCase				:      List[str]			  =				[hypothesis_template.format(_A) for x in candidate_labels]
     _UpperCAmelCase				:      Optional[Any]			  =				self.tokenizer(_A							,  return_tensors=self.framework							,  padding=_A)
     _UpperCAmelCase				:      str			  =				[text_inputs]
     return inputs


 def  snake_case__				(	self							,  _A) ->		List[str]:

     """simple docstring"""



     _UpperCAmelCase				:      Optional[int]			  =				model_inputs.pop('''candidate_labels''')
     _UpperCAmelCase				:      Union[str, Any]			  =				model_inputs.pop('''text_inputs''')
     if isinstance(text_inputs[0]							,  _A):
         _UpperCAmelCase				:      int			  =				text_inputs[0]
     else:
         # Batching case.
         _UpperCAmelCase				:      Dict			  =				text_inputs[0][0]

     _UpperCAmelCase				:      str			  =				self.model(**_A							,  **_A)

     _UpperCAmelCase				:      str			  =				{
         '''candidate_labels''': candidate_labels,
         '''logits''': outputs.logits_per_audio,
     }
     return model_outputs


 def  snake_case__				(	self							,  _A) ->		List[Any]:

     """simple docstring"""



     _UpperCAmelCase				:      str			  =				model_outputs.pop('''candidate_labels''')
     _UpperCAmelCase				:      str			  =				model_outputs['''logits'''][0]

     if self.framework == "pt":
         _UpperCAmelCase				:      int			  =				logits.softmax(dim=0)
         _UpperCAmelCase				:      Optional[Any]			  =				probs.tolist()
     else:
         raise ValueError('''`tf` framework not supported.''')

     _UpperCAmelCase				:      List[str]			  =				[
         {'''score''': score, '''label''': candidate_label}
         for score, candidate_label in sorted(zip(_A							,  _A)							,  key=lambda _A: -x[0])
     ]
     return result




