





import argparse
import logging
import os
import re

import tensorflow as tf

from transformers import (
    AutoConfig,
    AutoTokenizer,
    DataCollatorForLanguageModeling,
    PushToHubCallback,
    TFAutoModelForMaskedLM,
    create_optimizer,
)


_snake_case  :    List[str]						    =  logging.getLogger(__name__)

_snake_case  :    List[str]						    =  tf.data.AUTOTUNE

def   a_    ( ):
   __lowerCAmelCase								=	argparse.ArgumentParser(description='Train a masked language model on TPU.'	)
   parser.add_argument(
       '--pretrained_model_config',	type=lowerCAmelCase_,	default='roberta-base',	help='The model config to use. Note that we don\'t copy the model\'s weights, only the config!',	)
   parser.add_argument(
       '--tokenizer',	type=lowerCAmelCase_,	default='unigram-tokenizer-wikitext',	help='The name of the tokenizer to load. We use the pretrained tokenizer to initialize the model\'s vocab size.',	)

   parser.add_argument(
       '--per_replica_batch_size',	type=lowerCAmelCase_,	default=8,	help='Batch size per TPU core.',	)

   parser.add_argument(
       '--no_tpu',	action='store_true',	help='If set, run on CPU and don\'t try to initialize a TPU. Useful for debugging on non-TPU instances.',	)

   parser.add_argument(
       '--tpu_name',	type=lowerCAmelCase_,	help='Name of TPU resource to initialize. Should be blank on Colab, and \'local\' on TPU VMs.',	default='local',	)

   parser.add_argument(
       '--tpu_zone',	type=lowerCAmelCase_,	help='Google cloud zone that TPU resource is located in. Only used for non-Colab TPU nodes.',	)

   parser.add_argument(
       '--gcp_project',	type=lowerCAmelCase_,	help='Google cloud project name. Only used for non-Colab TPU nodes.'	)

   parser.add_argument(
       '--bfloat16',	action='store_true',	help='Use mixed-precision bfloat16 for training. This is the recommended lower-precision format for TPU.',	)

   parser.add_argument(
       '--train_dataset',	type=lowerCAmelCase_,	help='Path to training dataset to load. If the path begins with `gs://`'
       ' then the dataset will be loaded from a Google Cloud Storage bucket.',	)

   parser.add_argument(
       '--shuffle_buffer_size',	type=lowerCAmelCase_,	default=2**18,	help='Size of the shuffle buffer (in samples)',	)

   parser.add_argument(
       '--eval_dataset',	type=lowerCAmelCase_,	help='Path to evaluation dataset to load. If the path begins with `gs://`'
       ' then the dataset will be loaded from a Google Cloud Storage bucket.',	)

   parser.add_argument(
       '--num_epochs',	type=lowerCAmelCase_,	default=1,	help='Number of epochs to train for.',	)

   parser.add_argument(
       '--learning_rate',	type=lowerCAmelCase_,	default=1E-4,	help='Learning rate to use for training.',	)

   parser.add_argument(
       '--weight_decay_rate',	type=lowerCAmelCase_,	default=1E-3,	help='Weight decay rate to use for training.',	)

   parser.add_argument(
       '--max_length',	type=lowerCAmelCase_,	default=512,	help='Maximum length of tokenized sequences. Should match the setting used in prepare_tfrecord_shards.py',	)

   parser.add_argument(
       '--mlm_probability',	type=lowerCAmelCase_,	default=0.15,	help='Fraction of tokens to mask during training.',	)

   parser.add_argument('--output_dir',	type=lowerCAmelCase_,	required=lowerCAmelCase_,	help='Path to save model checkpoints to.'	)
   parser.add_argument('--hub_model_id',	type=lowerCAmelCase_,	help='Model ID to upload to on the Hugging Face Hub.'	)

   __lowerCAmelCase								=	parser.parse_args()
   return args

def   a_    ( lowerCAmelCase_      :			Union[str, Any]	):
   try:
      if args.tpu_name:
         __lowerCAmelCase								=	tf.distribute.cluster_resolver.TPUClusterResolver(
             args.tpu_name,	zone=args.tpu_zone,	project=args.gcp_project	)
      else:
         __lowerCAmelCase								=	tf.distribute.cluster_resolver.TPUClusterResolver()
   except ValueError:
      raise RuntimeError(
          'Couldn\'t connect to TPU! Most likely you need to specify --tpu_name, --tpu_zone, or '
          '--gcp_project. When running on a TPU VM, use --tpu_name local.'	)

   tf.config.experimental_connect_to_cluster(lowerCAmelCase_	)
   tf.tpu.experimental.initialize_tpu_system(lowerCAmelCase_	)

   return tpu

def   a_    ( lowerCAmelCase_      :			Dict	):
   __lowerCAmelCase								=	0
   for file in file_list:
      __lowerCAmelCase								=	file.split('/'	)[-1]
      __lowerCAmelCase								=	re.search(R'-\d+-(\d+)\.tfrecord',	lowerCAmelCase_	).group(1	)
      __lowerCAmelCase								=	int(lowerCAmelCase_	)
      num_samples += sample_count

   return num_samples

def   a_    ( lowerCAmelCase_      :			Optional[int],	lowerCAmelCase_      :			List[Any],	lowerCAmelCase_      :			Tuple,	lowerCAmelCase_      :			Optional[Any],	lowerCAmelCase_      :			Tuple,	lowerCAmelCase_      :			List[str]=None	):
   __lowerCAmelCase								=	count_samples(lowerCAmelCase_	)
   __lowerCAmelCase								=	tf.data.Dataset.from_tensor_slices(lowerCAmelCase_	)
   if shuffle:
      __lowerCAmelCase								=	dataset.shuffle(len(lowerCAmelCase_	)	)
   __lowerCAmelCase								=	tf.data.TFRecordDataset(lowerCAmelCase_,	num_parallel_reads=lowerCAmelCase_	)
   # TF can't infer the total sample count because it doesn't read all the records yet, so we assert it here
   __lowerCAmelCase								=	dataset.apply(tf.data.experimental.assert_cardinality(lowerCAmelCase_	)	)
   __lowerCAmelCase								=	dataset.map(lowerCAmelCase_,	num_parallel_calls=lowerCAmelCase_	)
   if shuffle:
      assert shuffle_buffer_size is not None
      __lowerCAmelCase								=	dataset.shuffle(args.shuffle_buffer_size	)
   __lowerCAmelCase								=	dataset.batch(lowerCAmelCase_,	drop_remainder=lowerCAmelCase_	)
   __lowerCAmelCase								=	dataset.map(lowerCAmelCase_,	num_parallel_calls=lowerCAmelCase_	)
   __lowerCAmelCase								=	dataset.prefetch(lowerCAmelCase_	)
   return dataset

def   a_    ( lowerCAmelCase_      :			Optional[Any]	):
   if not args.no_tpu:
      __lowerCAmelCase								=	initialize_tpu(lowerCAmelCase_	)
      __lowerCAmelCase								=	tf.distribute.TPUStrategy(lowerCAmelCase_	)
   else:
      __lowerCAmelCase								=	tf.distribute.OneDeviceStrategy(device='/gpu:0'	)

   if args.bfloataa:
      tf.keras.mixed_precision.set_global_policy('mixed_bfloat16'	)

   __lowerCAmelCase								=	AutoTokenizer.from_pretrained(args.tokenizer	)
   __lowerCAmelCase								=	AutoConfig.from_pretrained(args.pretrained_model_config	)
   __lowerCAmelCase								=	tokenizer.vocab_size

   __lowerCAmelCase								=	tf.io.gfile.glob(os.path.join(args.train_dataset,	'*.tfrecord'	)	)
   if not training_records:
      raise ValueError(F"""No .tfrecord files found in {args.train_dataset}."""	)
   __lowerCAmelCase								=	tf.io.gfile.glob(os.path.join(args.eval_dataset,	'*.tfrecord'	)	)
   if not eval_records:
      raise ValueError(F"""No .tfrecord files found in {args.eval_dataset}."""	)

   __lowerCAmelCase								=	count_samples(lowerCAmelCase_	)

   __lowerCAmelCase								=	num_train_samples // (args.per_replica_batch_size * strategy.num_replicas_in_sync)
   __lowerCAmelCase								=	steps_per_epoch * args.num_epochs

   with strategy.scope():
      __lowerCAmelCase								=	TFAutoModelForMaskedLM.from_config(lowerCAmelCase_	)
      model(model.dummy_inputs	)  # Pass some dummy inputs through the model to ensure all the weights are built
      __lowerCAmelCase	,	__lowerCAmelCase								=	create_optimizer(
          num_train_steps=lowerCAmelCase_,	num_warmup_steps=total_train_steps // 20,	init_lr=args.learning_rate,	weight_decay_rate=args.weight_decay_rate,	)

      # Transformers models compute the right loss for their task by default when labels are passed, and will
      # use this for training unless you specify your own loss function in compile().
      model.compile(optimizer=lowerCAmelCase_,	metrics=['accuracy']	)

   def decode_fn(lowerCAmelCase_      :			List[Any]	):
      __lowerCAmelCase								=	{
          'input_ids': tf.io.FixedLenFeature(dtype=tf.intaa,	shape=(args.max_length,)	),
          'attention_mask': tf.io.FixedLenFeature(dtype=tf.intaa,	shape=(args.max_length,)	),
      }
      return tf.io.parse_single_example(lowerCAmelCase_,	lowerCAmelCase_	)

   # Many of the data collators in Transformers are TF-compilable when return_tensors == "tf", so we can
   # use their methods in our data pipeline.
   __lowerCAmelCase								=	DataCollatorForLanguageModeling(
       tokenizer=lowerCAmelCase_,	mlm_probability=args.mlm_probability,	mlm=lowerCAmelCase_,	return_tensors='tf'	)

   def mask_with_collator(lowerCAmelCase_      :			Union[str, Any]	):
      # TF really needs an isin() function
      __lowerCAmelCase								=	(
          ~tf.cast(batch['attention_mask'],	tf.bool	)
          | (batch['input_ids'] == tokenizer.cls_token_id)
          | (batch['input_ids'] == tokenizer.sep_token_id)
      )
      __lowerCAmelCase	,	__lowerCAmelCase								=	data_collator.tf_mask_tokens(
          batch['input_ids'],	vocab_size=len(lowerCAmelCase_	),	mask_token_id=tokenizer.mask_token_id,	special_tokens_mask=lowerCAmelCase_,	)
      return batch

   __lowerCAmelCase								=	args.per_replica_batch_size * strategy.num_replicas_in_sync

   __lowerCAmelCase								=	prepare_dataset(
       lowerCAmelCase_,	decode_fn=lowerCAmelCase_,	mask_fn=lowerCAmelCase_,	batch_size=lowerCAmelCase_,	shuffle=lowerCAmelCase_,	shuffle_buffer_size=args.shuffle_buffer_size,	)

   __lowerCAmelCase								=	prepare_dataset(
       lowerCAmelCase_,	decode_fn=lowerCAmelCase_,	mask_fn=lowerCAmelCase_,	batch_size=lowerCAmelCase_,	shuffle=lowerCAmelCase_,	)

   __lowerCAmelCase								=	[]
   if args.hub_model_id:
      callbacks.append(
          PushToHubCallback(output_dir=args.output_dir,	hub_model_id=args.hub_model_id,	tokenizer=lowerCAmelCase_	)	)

   model.fit(
       lowerCAmelCase_,	validation_data=lowerCAmelCase_,	epochs=args.num_epochs,	callbacks=lowerCAmelCase_,	)

   model.save_pretrained(args.output_dir	)


if __name__ == "__main__":
     _snake_case  :    Union[str, Any]						    =  parse_args()
     main(args)








import pytest

from datasets.utils.sharding import _distribute_shards, _number_of_shards_in_gen_kwargs, _split_gen_kwargs

@pytest.mark.parametrize(
    'kwargs, expected',	[
        ({'num_shards': 0, 'max_num_jobs': 1}, []),
        ({'num_shards': 10, 'max_num_jobs': 1}, [range(10	)]),
        ({'num_shards': 10, 'max_num_jobs': 10}, [range(lowerCAmelCase_,	i + 1	) for i in range(10	)]),
        ({'num_shards': 1, 'max_num_jobs': 10}, [range(1	)]),
        ({'num_shards': 10, 'max_num_jobs': 3}, [range(0,	4	), range(4,	7	), range(7,	10	)]),
        ({'num_shards': 3, 'max_num_jobs': 10}, [range(0,	1	), range(1,	2	), range(2,	3	)]),
    ],	)
def   a_    ( lowerCAmelCase_      :			Optional[Any],	lowerCAmelCase_      :			Any	):
   __lowerCAmelCase								=	_distribute_shards(**lowerCAmelCase_	)
   assert out == expected

@pytest.mark.parametrize(
    'gen_kwargs, max_num_jobs, expected',	[
        ({'foo': 0}, 10, [{'foo': 0}]),
        ({'shards': [0, 1, 2, 3]}, 1, [{'shards': [0, 1, 2, 3]}]),
        ({'shards': [0, 1, 2, 3]}, 4, [{'shards': [0]}, {'shards': [1]}, {'shards': [2]}, {'shards': [3]}]),
        ({'shards': [0, 1]}, 4, [{'shards': [0]}, {'shards': [1]}]),
        ({'shards': [0, 1, 2, 3]}, 2, [{'shards': [0, 1]}, {'shards': [2, 3]}]),
    ],	)
def   a_    ( lowerCAmelCase_      :			Optional[Any],	lowerCAmelCase_      :			List[str],	lowerCAmelCase_      :			Optional[int]	):
   __lowerCAmelCase								=	_split_gen_kwargs(lowerCAmelCase_,	lowerCAmelCase_	)
   assert out == expected




@pytest.mark.parametrize(
    'gen_kwargs, expected',	[
        ({'foo': 0}, 1),
        ({'shards': [0]}, 1),
        ({'shards': [0, 1, 2, 3]}, 4),
        ({'shards': [0, 1, 2, 3], 'foo': 0}, 4),
        ({'shards': [0, 1, 2, 3], 'other': (0, 1)}, 4),
        ({'shards': [0, 1, 2, 3], 'shards2': [0, 1]}, RuntimeError),
    ],	)
def   a_    ( lowerCAmelCase_      :			int,	lowerCAmelCase_      :			Any	):
   if expected is RuntimeError:
      with pytest.raises(lowerCAmelCase_	):
         _number_of_shards_in_gen_kwargs(lowerCAmelCase_	)
   else:
      __lowerCAmelCase								=	_number_of_shards_in_gen_kwargs(lowerCAmelCase_	)
      assert out == expected

