





"""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	     =					logging.getLogger(__name__)

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

				logger.info(F"""Loading Tokenizer ({args.tokenizer_name})""")
				if args.tokenizer_type == "bert":
								__a	:  Dict         =					BertTokenizer.from_pretrained(args.tokenizer_name)
								__a	:  Any         =					tokenizer.special_tokens_map['''cls_token''']  # `[CLS]`
								__a	:  Union[str, Any]         =					tokenizer.special_tokens_map['''sep_token''']  # `[SEP]`
				elif args.tokenizer_type == "roberta":
								__a	:  int         =					RobertaTokenizer.from_pretrained(args.tokenizer_name)
								__a	:  Optional[Any]         =					tokenizer.special_tokens_map['''cls_token''']  # `<s>`
								__a	:  Tuple         =					tokenizer.special_tokens_map['''sep_token''']  # `</s>`
				elif args.tokenizer_type == "gpt2":
								__a	:  Optional[int]         =					GPTaTokenizer.from_pretrained(args.tokenizer_name)
								__a	:  Union[str, Any]         =					tokenizer.special_tokens_map['''bos_token''']  # `<|endoftext|>`
								__a	:  Tuple         =					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:
								__a	:  Any         =					fp.readlines()

				logger.info('''Start encoding''')
				logger.info(F"""{len(a_)} examples to process.""")

				__a	:  List[Any]         =					[]
				__a	:  Any         =					0
				__a	:  List[Any]         =					1_00_00
				__a	:  Tuple         =					time.time()
				for text in data:
								__a	:  List[Any]         =					F"""{bos} {text.strip()} {sep}"""
								__a	:  Any         =					tokenizer.encode(a_				,    add_special_tokens=a_)
								rslt.append(a_)

								iter += 1
								if iter % interval == 0:
												__a	:  int         =					time.time()
												logger.info(F"""{iter} examples processed. - {(end-start):.2f}s/{interval}expl""")
												__a	:  Optional[int]         =					time.time()
				logger.info('''Finished binarization''')
				logger.info(F"""{len(a_)} examples processed.""")

				__a	:  int         =					F"""{args.dump_file}.{args.tokenizer_name}.pickle"""
				__a	:  List[Any]         =					tokenizer.vocab_size
				if vocab_size < (1 << 16):
								__a	:  Union[str, Any]         =					[np.uintaa(a_) for d in rslt]
				else:
								__a	:  Dict         =					[np.intaa(a_) for d in rslt]
				random.shuffle(rslt_)
				logger.info(F"""Dump to {dp_file}""")
				with open(a_				,    '''wb''') as handle:
								pickle.dump(rslt_				,    a_				,    protocol=pickle.HIGHEST_PROTOCOL)


if __name__ == "__main__":
		main()






"""simple docstring"""






# Copyright 2023 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from ..models.whisper import WhisperForConditionalGeneration, WhisperProcessor
from .base import PipelineTool




class 			__lowercase     (     _UpperCamelCase							):







					'''simple docstring'''




					__lowerCAmelCase					=					'''openai/whisper-base'''
					__lowerCAmelCase					=					(
					    '''This is a tool that transcribes an audio into text. It takes an input named `audio` and returns the '''
					    '''transcribed text.'''
					)
					__lowerCAmelCase					=					'''transcriber'''
					__lowerCAmelCase					=					WhisperProcessor
					__lowerCAmelCase					=					WhisperForConditionalGeneration

					__lowerCAmelCase					=					['''audio''']
					__lowerCAmelCase					=					['''text''']







					def 	_lowerCamelCase						(  self		,     _UpperCAmelCase       ):
									return self.pre_processor(_UpperCAmelCase		,     return_tensors='''pt'''       ).input_features







					def 	_lowerCamelCase						(  self		,     _UpperCAmelCase       ):
									return self.model.generate(inputs=_UpperCAmelCase       )







					def 	_lowerCamelCase						(  self		,     _UpperCAmelCase       ):
									return self.pre_processor.batch_decode(_UpperCAmelCase		,     skip_special_tokens=_UpperCAmelCase       )[0]