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



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

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

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

							lowerCamelCase_					=       []
							lowerCamelCase_					=       0
							lowerCamelCase_					=       1_0_0_0_0
							lowerCamelCase_					=       time.time()
							for text in data:
														lowerCamelCase_					=       F'''{bos} {text.strip()} {sep}'''
														lowerCamelCase_					=       tokenizer.encode(lowerCAmelCase__      ,					add_special_tokens=lowerCAmelCase__							)
														rslt.append(lowerCAmelCase__							)

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

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


if __name__ == "__main__":
		main()










import unittest

from diffusers.pipelines.pipeline_utils import is_safetensors_compatible



class        lowerCAmelCase__					(  unittest.TestCase       ):


				def __UpperCamelCase						(							self      :  List[str]       )		->     Any:
											A      						=    [
											    'safety_checker/pytorch_model.bin',
											    'safety_checker/model.safetensors',
											    'vae/diffusion_pytorch_model.bin',
											    'vae/diffusion_pytorch_model.safetensors',
											    'text_encoder/pytorch_model.bin',
											    'text_encoder/model.safetensors',
											    'unet/diffusion_pytorch_model.bin',
											    'unet/diffusion_pytorch_model.safetensors',
											]
											self.assertTrue(is_safetensors_compatible(__UpperCamelCase       )       )


				def __UpperCamelCase						(							self      :  str       )		->     List[Any]:
											A      						=    [
											    'unet/diffusion_pytorch_model.bin',
											    'unet/diffusion_pytorch_model.safetensors',
											]
											self.assertTrue(is_safetensors_compatible(__UpperCamelCase       )       )


				def __UpperCamelCase						(							self      :  Any       )		->     Tuple:
											A      						=    [
											    'safety_checker/pytorch_model.bin',
											    'safety_checker/model.safetensors',
											    'vae/diffusion_pytorch_model.bin',
											    'vae/diffusion_pytorch_model.safetensors',
											    'text_encoder/pytorch_model.bin',
											    'text_encoder/model.safetensors',
											    'unet/diffusion_pytorch_model.bin',
											    # Removed: 'unet/diffusion_pytorch_model.safetensors',
											]
											self.assertFalse(is_safetensors_compatible(__UpperCamelCase       )       )


				def __UpperCamelCase						(							self      :  Dict       )		->     Any:
											A      						=    [
											    'text_encoder/pytorch_model.bin',
											    'text_encoder/model.safetensors',
											]
											self.assertTrue(is_safetensors_compatible(__UpperCamelCase       )       )


				def __UpperCamelCase						(							self      :  Optional[int]       )		->     List[str]:
											A      						=    [
											    'safety_checker/pytorch_model.bin',
											    'safety_checker/model.safetensors',
											    'vae/diffusion_pytorch_model.bin',
											    'vae/diffusion_pytorch_model.safetensors',
											    'text_encoder/pytorch_model.bin',
											    # Removed: 'text_encoder/model.safetensors',
											    'unet/diffusion_pytorch_model.bin',
											    'unet/diffusion_pytorch_model.safetensors',
											]
											self.assertFalse(is_safetensors_compatible(__UpperCamelCase       )       )


				def __UpperCamelCase						(							self      :  List[str]       )		->     List[str]:
											A      						=    [
											    'safety_checker/pytorch_model.fp16.bin',
											    'safety_checker/model.fp16.safetensors',
											    'vae/diffusion_pytorch_model.fp16.bin',
											    'vae/diffusion_pytorch_model.fp16.safetensors',
											    'text_encoder/pytorch_model.fp16.bin',
											    'text_encoder/model.fp16.safetensors',
											    'unet/diffusion_pytorch_model.fp16.bin',
											    'unet/diffusion_pytorch_model.fp16.safetensors',
											]
											A      						=    'fp16'
											self.assertTrue(is_safetensors_compatible(__UpperCamelCase						, variant=__UpperCamelCase       )       )


				def __UpperCamelCase						(							self      :  Optional[Any]       )		->     Optional[Any]:
											A      						=    [
											    'unet/diffusion_pytorch_model.fp16.bin',
											    'unet/diffusion_pytorch_model.fp16.safetensors',
											]
											A      						=    'fp16'
											self.assertTrue(is_safetensors_compatible(__UpperCamelCase						, variant=__UpperCamelCase       )       )


				def __UpperCamelCase						(							self      :  int       )		->     Optional[int]:
											# pass variant but use the non-variant filenames
											A      						=    [
											    'unet/diffusion_pytorch_model.bin',
											    'unet/diffusion_pytorch_model.safetensors',
											]
											A      						=    'fp16'
											self.assertTrue(is_safetensors_compatible(__UpperCamelCase						, variant=__UpperCamelCase       )       )


				def __UpperCamelCase						(							self      :  Any       )		->     List[str]:
											A      						=    [
											    'safety_checker/pytorch_model.fp16.bin',
											    'safety_checker/model.fp16.safetensors',
											    'vae/diffusion_pytorch_model.fp16.bin',
											    'vae/diffusion_pytorch_model.fp16.safetensors',
											    'text_encoder/pytorch_model.fp16.bin',
											    'text_encoder/model.fp16.safetensors',
											    'unet/diffusion_pytorch_model.fp16.bin',
											    # Removed: 'unet/diffusion_pytorch_model.fp16.safetensors',
											]
											A      						=    'fp16'
											self.assertFalse(is_safetensors_compatible(__UpperCamelCase						, variant=__UpperCamelCase       )       )


				def __UpperCamelCase						(							self      :  Optional[Any]       )		->     Optional[int]:
											A      						=    [
											    'text_encoder/pytorch_model.fp16.bin',
											    'text_encoder/model.fp16.safetensors',
											]
											A      						=    'fp16'
											self.assertTrue(is_safetensors_compatible(__UpperCamelCase						, variant=__UpperCamelCase       )       )


				def __UpperCamelCase						(							self      :  List[str]       )		->     int:
											# pass variant but use the non-variant filenames
											A      						=    [
											    'text_encoder/pytorch_model.bin',
											    'text_encoder/model.safetensors',
											]
											A      						=    'fp16'
											self.assertTrue(is_safetensors_compatible(__UpperCamelCase						, variant=__UpperCamelCase       )       )


				def __UpperCamelCase						(							self      :  List[str]       )		->     Optional[Any]:
											A      						=    [
											    'safety_checker/pytorch_model.fp16.bin',
											    'safety_checker/model.fp16.safetensors',
											    'vae/diffusion_pytorch_model.fp16.bin',
											    'vae/diffusion_pytorch_model.fp16.safetensors',
											    'text_encoder/pytorch_model.fp16.bin',
											    # 'text_encoder/model.fp16.safetensors',
											    'unet/diffusion_pytorch_model.fp16.bin',
											    'unet/diffusion_pytorch_model.fp16.safetensors',
											]
											A      						=    'fp16'
											self.assertFalse(is_safetensors_compatible(__UpperCamelCase						, variant=__UpperCamelCase       )       )