






# 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     _SCREAMING_SNAKE_CASE       (			_a      ):
	snake_case__							:     Any     	=  """openai/whisper-base"""
	snake_case__							:     Optional[int]     	=  (
	    """This is a tool that transcribes an audio into text. It takes an input named `audio` and returns the """
	    """transcribed text."""
	)
	snake_case__							:     Any     	=  """transcriber"""
	snake_case__							:     Optional[int]     	=  WhisperProcessor
	snake_case__							:     str     	=  WhisperForConditionalGeneration

	snake_case__							:     Optional[Any]     	=  ["""audio"""]
	snake_case__							:     Any     	=  ["""text"""]





	def 						_A (		self				:					str ,  __lowerCamelCase				:					Dict	):
		return self.pre_processor(__lowerCamelCase ,  return_tensors="""pt"""	).input_features





	def 						_A (		self				:					Dict ,  __lowerCamelCase				:					List[Any]	):
		return self.model.generate(inputs=__lowerCamelCase	)





	def 						_A (		self				:					Any ,  __lowerCamelCase				:					Optional[Any]	):
		return self.pre_processor.batch_decode(__lowerCamelCase ,  skip_special_tokens=__lowerCamelCase	)[0]







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



def      __lowerCamelCase   (							)			->      int:
      __SCREAMING_SNAKE_CASE   :Union[str, Any]						=  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.'''							)
      __SCREAMING_SNAKE_CASE   :Any						=  parser.parse_args()

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

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

      __SCREAMING_SNAKE_CASE   :Optional[int]						=  []
      __SCREAMING_SNAKE_CASE   :List[str]						=  0
      __SCREAMING_SNAKE_CASE   :Optional[Any]						=  1_00_00
      __SCREAMING_SNAKE_CASE   :List[Any]						=  time.time()
      for text in data:
            __SCREAMING_SNAKE_CASE   :Any						=  f'''{bos} {text.strip()} {sep}'''
            __SCREAMING_SNAKE_CASE   :int						=  tokenizer.encode(a_ ,   add_special_tokens=a_							)
            rslt.append(a_							)

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

      __SCREAMING_SNAKE_CASE   :Optional[int]						=  f'''{args.dump_file}.{args.tokenizer_name}.pickle'''
      __SCREAMING_SNAKE_CASE   :str						=  tokenizer.vocab_size
      if vocab_size < (1 << 16):
            __SCREAMING_SNAKE_CASE   :Union[str, Any]						=  [np.uintaa(a_							) for d in rslt]
      else:
            __SCREAMING_SNAKE_CASE   :List[Any]						=  [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()