



"""simple docstring"""


import argparse
import logging
import pickle
import random
import time

import numpy as np

from transformers import BertTokenizer, GPTaTokenizer, RobertaTokenizer


logging.basicConfig(
    format="%(asctime)s - %(levelname)s - %(name)s - %(message)s", datefmt="%m/%d/%Y %H:%M:%S", level=logging.INFO
)
__A   :							Union[str, Any]										=   logging.getLogger(__name__)







def  lowercase	(      ):






				'''simple docstring'''



				_UpperCAmelCase					=    argparse.ArgumentParser(
				    description='''Preprocess the data to avoid re-doing it several times by (tokenization + token_to_ids).'''				)
				parser.add_argument('''--file_path'''					,     type=_SCREAMING_SNAKE_CASE					,     default='''data/dump.txt'''					,     help='''The path to the data.'''				)
				parser.add_argument('''--tokenizer_type'''					,     type=_SCREAMING_SNAKE_CASE					,     default='''bert'''					,     choices=['''bert''', '''roberta''', '''gpt2''']				)
				parser.add_argument('''--tokenizer_name'''					,     type=_SCREAMING_SNAKE_CASE					,     default='''bert-base-uncased'''					,     help='''The tokenizer to use.'''				)
				parser.add_argument('''--dump_file'''					,     type=_SCREAMING_SNAKE_CASE					,     default='''data/dump'''					,     help='''The dump file prefix.'''				)
				_UpperCAmelCase					=    parser.parse_args()

				logger.info(f'Loading Tokenizer ({args.tokenizer_name})'				)
				if args.tokenizer_type == "bert":
								_UpperCAmelCase					=    BertTokenizer.from_pretrained(args.tokenizer_name				)
								_UpperCAmelCase					=    tokenizer.special_tokens_map['''cls_token''']  # `[CLS]`
								_UpperCAmelCase					=    tokenizer.special_tokens_map['''sep_token''']  # `[SEP]`
				elif args.tokenizer_type == "roberta":
								_UpperCAmelCase					=    RobertaTokenizer.from_pretrained(args.tokenizer_name				)
								_UpperCAmelCase					=    tokenizer.special_tokens_map['''cls_token''']  # `<s>`
								_UpperCAmelCase					=    tokenizer.special_tokens_map['''sep_token''']  # `</s>`
				elif args.tokenizer_type == "gpt2":
								_UpperCAmelCase					=    GPTaTokenizer.from_pretrained(args.tokenizer_name				)
								_UpperCAmelCase					=    tokenizer.special_tokens_map['''bos_token''']  # `<|endoftext|>`
								_UpperCAmelCase					=    tokenizer.special_tokens_map['''eos_token''']  # `<|endoftext|>`

				logger.info(f'Loading text from {args.file_path}'				)
				with open(args.file_path					,     '''r'''					,     encoding='''utf8'''				) as fp:
								_UpperCAmelCase					=    fp.readlines()

				logger.info('''Start encoding'''				)
				logger.info(f'{len(_SCREAMING_SNAKE_CASE				)} examples to process.'				)

				_UpperCAmelCase					=    []
				_UpperCAmelCase					=    0
				_UpperCAmelCase					=    1_0000
				_UpperCAmelCase					=    time.time()
				for text in data:
								_UpperCAmelCase					=    f'{bos} {text.strip()} {sep}'
								_UpperCAmelCase					=    tokenizer.encode(_SCREAMING_SNAKE_CASE					,     add_special_tokens=_SCREAMING_SNAKE_CASE				)
								rslt.append(_SCREAMING_SNAKE_CASE				)

								iter += 1
								if iter % interval == 0:
												_UpperCAmelCase					=    time.time()
												logger.info(f'{iter} examples processed. - {(end-start):.2f}s/{interval}expl'				)
												_UpperCAmelCase					=    time.time()
				logger.info('''Finished binarization'''				)
				logger.info(f'{len(_SCREAMING_SNAKE_CASE				)} examples processed.'				)

				_UpperCAmelCase					=    f'{args.dump_file}.{args.tokenizer_name}.pickle'
				_UpperCAmelCase					=    tokenizer.vocab_size
				if vocab_size < (1 << 16):
								_UpperCAmelCase					=    [np.uintaa(_SCREAMING_SNAKE_CASE				) for d in rslt]
				else:
								_UpperCAmelCase					=    [np.intaa(_SCREAMING_SNAKE_CASE				) for d in rslt]
				random.shuffle(rslt_				)
				logger.info(f'Dump to {dp_file}'				)
				with open(_SCREAMING_SNAKE_CASE					,     '''wb'''				) as handle:
								pickle.dump(rslt_					,     _SCREAMING_SNAKE_CASE					,     protocol=pickle.HIGHEST_PROTOCOL				)


if __name__ == "__main__":
					main()








"""simple docstring"""


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


__A   :							List[str]										=   logging.get_logger(__name__)
@add_end_docstrings(lowerCAmelCase)
class 			_a (					lowerCAmelCase):
			"""simple docstring"""



			def __init__(		self     :    Union[str, Any]				,   **__UpperCamelCase     :    Optional[Any]   )->List[Any]:
							super().__init__(**__UpperCamelCase   )

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



			def __call__(		self     :    Optional[int]				,   __UpperCamelCase     :    Union[np.ndarray, bytes, str]				,   **__UpperCamelCase     :    Tuple   )->List[str]:
							return super().__call__(__UpperCamelCase				,   **__UpperCamelCase   )



			def  lowercase__						(		self     :    Any				,   **__UpperCamelCase     :    Any   )->Union[str, Any]:
							_UpperCAmelCase					=    {}
							if "candidate_labels" in kwargs:
											_UpperCAmelCase					=    kwargs['''candidate_labels''']
							if "hypothesis_template" in kwargs:
											_UpperCAmelCase					=    kwargs['''hypothesis_template''']

							return preprocess_params, {}, {}



			def  lowercase__						(		self     :    Optional[int]				,   __UpperCamelCase     :    Union[str, Any]				,   __UpperCamelCase     :    List[Any]=None				,   __UpperCamelCase     :    Tuple="This is a sound of {}."   )->int:
							if isinstance(__UpperCamelCase				,   __UpperCamelCase   ):
											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					=    requests.get(__UpperCamelCase   ).content
											else:
															with open(__UpperCamelCase				,   '''rb'''   ) as f:
																			_UpperCAmelCase					=    f.read()

							if isinstance(__UpperCamelCase				,   __UpperCamelCase   ):
											_UpperCAmelCase					=    ffmpeg_read(__UpperCamelCase				,   self.feature_extractor.sampling_rate   )

							if not isinstance(__UpperCamelCase				,   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					=    self.feature_extractor(
							    [audio]				,   sampling_rate=self.feature_extractor.sampling_rate				,   return_tensors='''pt'''   )
							_UpperCAmelCase					=    candidate_labels
							_UpperCAmelCase					=    [hypothesis_template.format(__UpperCamelCase   ) for x in candidate_labels]
							_UpperCAmelCase					=    self.tokenizer(__UpperCamelCase				,   return_tensors=self.framework				,   padding=__UpperCamelCase   )
							_UpperCAmelCase					=    [text_inputs]
							return inputs



			def  lowercase__						(		self     :    Union[str, Any]				,   __UpperCamelCase     :    Optional[int]   )->Any:
							_UpperCAmelCase					=    model_inputs.pop('''candidate_labels'''   )
							_UpperCAmelCase					=    model_inputs.pop('''text_inputs'''   )
							if isinstance(text_inputs[0]				,   __UpperCamelCase   ):
											_UpperCAmelCase					=    text_inputs[0]
							else:
											# Batching case.
											_UpperCAmelCase					=    text_inputs[0][0]

							_UpperCAmelCase					=    self.model(**__UpperCamelCase				,   **__UpperCamelCase   )

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



			def  lowercase__						(		self     :    List[str]				,   __UpperCamelCase     :    Any   )->List[Any]:
							_UpperCAmelCase					=    model_outputs.pop('''candidate_labels'''   )
							_UpperCAmelCase					=    model_outputs['''logits'''][0]

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

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



