






"""simple docstring"""





import os
from typing import List, Union

import tensorflow as tf
from tensorflow_text import BertTokenizer as BertTokenizerLayer
from tensorflow_text import FastBertTokenizer, ShrinkLongestTrimmer, case_fold_utfa, combine_segments, pad_model_inputs

from .tokenization_bert import BertTokenizer







class 						_a  (		tf.keras.layers.Layer       ):





   '''simple docstring'''






   def __init__(       self					, UpperCAmelCase_					, UpperCAmelCase_					, UpperCAmelCase_ = None					, UpperCAmelCase_ = None					, UpperCAmelCase_ = None					, UpperCAmelCase_ = "longest"					, UpperCAmelCase_ = True					, UpperCAmelCase_ = 512					, UpperCAmelCase_ = None					, UpperCAmelCase_ = True					, UpperCAmelCase_ = True					, UpperCAmelCase_ = True					, **UpperCAmelCase_					, )			->   str:




          '''simple docstring'''




          super().__init__()
          if use_fast_bert_tokenizer:
                 lowercase__:							List[str]         =     FastBertTokenizer(
                     UpperCamelCase__					, token_out_type=tf.intaa					, lower_case_nfd_strip_accents=UpperCamelCase__					, **UpperCamelCase__)
          else:
                 lowercase__:							Any         =     tf.lookup.StaticVocabularyTable(
                     tf.lookup.KeyValueTensorInitializer(
                         keys=UpperCamelCase__					, key_dtype=tf.string					, values=tf.range(tf.size(UpperCamelCase__					, out_type=tf.intaa)					, dtype=tf.intaa)					, value_dtype=tf.intaa					, )					, num_oov_buckets=1					, )
                 lowercase__:							Optional[int]         =     BertTokenizerLayer(
                     UpperCamelCase__					, token_out_type=tf.intaa					, lower_case=UpperCamelCase__					, **UpperCamelCase__)

          lowercase__:							Union[str, Any]         =     vocab_list
          lowercase__:							Dict         =     do_lower_case
          lowercase__:							List[str]         =     cls_token_id or vocab_list.index("[CLS]")
          lowercase__:							Dict         =     sep_token_id or vocab_list.index("[SEP]")
          lowercase__:							Tuple         =     pad_token_id or vocab_list.index("[PAD]")
          lowercase__:							Dict         =     ShrinkLongestTrimmer(max_length - 3					, axis=1)  # Allow room for special tokens
          lowercase__:							List[Any]         =     max_length
          lowercase__:							str         =     padding
          lowercase__:							Any         =     truncation
          lowercase__:							Tuple         =     pad_to_multiple_of
          lowercase__:							List[str]         =     return_token_type_ids
          lowercase__:							Any         =     return_attention_mask





   @classmethod
   def __lowercase							(       cls					, UpperCAmelCase_					, **UpperCAmelCase_)			->   Any:  # noqa: F821




          '''simple docstring'''




          lowercase__:							Optional[int]         =     kwargs.pop("do_lower_case"					, UpperCamelCase__)
          lowercase__:							int         =     tokenizer.do_lower_case if do_lower_case is None else do_lower_case
          lowercase__:							str         =     kwargs.pop("cls_token_id"					, UpperCamelCase__)
          lowercase__:							Optional[int]         =     tokenizer.cls_token_id if cls_token_id is None else cls_token_id
          lowercase__:							Dict         =     kwargs.pop("sep_token_id"					, UpperCamelCase__)
          lowercase__:							List[Any]         =     tokenizer.sep_token_id if sep_token_id is None else sep_token_id
          lowercase__:							List[str]         =     kwargs.pop("pad_token_id"					, UpperCamelCase__)
          lowercase__:							Dict         =     tokenizer.pad_token_id if pad_token_id is None else pad_token_id

          lowercase__:							List[Any]         =     tokenizer.get_vocab()
          lowercase__:							Dict         =     sorted([(wordpiece, idx) for wordpiece, idx in vocab.items()]					, key=lambda UpperCAmelCase_: x[1])
          lowercase__:							Union[str, Any]         =     [entry[0] for entry in vocab]
          return cls(
              vocab_list=UpperCamelCase__					, do_lower_case=UpperCamelCase__					, cls_token_id=UpperCamelCase__					, sep_token_id=UpperCamelCase__					, pad_token_id=UpperCamelCase__					, **UpperCamelCase__					, )





   @classmethod
   def __lowercase							(       cls					, UpperCAmelCase_					, *UpperCAmelCase_					, **UpperCAmelCase_)			->   str:




          '''simple docstring'''




          try:
                 lowercase__:							Dict         =     BertTokenizer.from_pretrained(UpperCamelCase__					, *UpperCamelCase__					, **UpperCamelCase__)
          except:  # noqa: E722
                 from .tokenization_bert_fast import BertTokenizerFast

                 lowercase__:							Optional[Any]         =     BertTokenizerFast.from_pretrained(UpperCamelCase__					, *UpperCamelCase__					, **UpperCamelCase__)
          return cls.from_tokenizer(UpperCamelCase__					, **UpperCamelCase__)





   def __lowercase							(       self					, UpperCAmelCase_)			->   Dict:




          '''simple docstring'''




          if self.do_lower_case:
                 lowercase__:							str         =     case_fold_utfa(UpperCamelCase__)
          lowercase__:							Optional[int]         =     self.tf_tokenizer.tokenize(UpperCamelCase__)
          return tokens.merge_dims(1					, -1)





   def __lowercase							(       self					, UpperCAmelCase_					, UpperCAmelCase_=None					, UpperCAmelCase_=None					, UpperCAmelCase_=None					, UpperCAmelCase_=None					, UpperCAmelCase_=None					, UpperCAmelCase_=None					, UpperCAmelCase_=None					, )			->   Any:




          '''simple docstring'''




          if padding is None:
                 lowercase__:							str         =     self.padding
          if padding not in ("longest", "max_length"):
                 raise ValueError("Padding must be either \'longest\' or \'max_length\'!")
          if max_length is not None and text_pair is not None:
                 # Because we have to instantiate a Trimmer to do it properly
                 raise ValueError("max_length cannot be overridden at call time when truncating paired texts!")
          if max_length is None:
                 lowercase__:							List[Any]         =     self.max_length
          if truncation is None:
                 lowercase__:							Optional[int]         =     self.truncation
          if pad_to_multiple_of is None:
                 lowercase__:							Optional[int]         =     self.pad_to_multiple_of
          if return_token_type_ids is None:
                 lowercase__:							Union[str, Any]         =     self.return_token_type_ids
          if return_attention_mask is None:
                 lowercase__:							Optional[int]         =     self.return_attention_mask
          if not isinstance(UpperCamelCase__					, tf.Tensor):
                 lowercase__:							Union[str, Any]         =     tf.convert_to_tensor(UpperCamelCase__)
          if text_pair is not None and not isinstance(UpperCamelCase__					, tf.Tensor):
                 lowercase__:							List[str]         =     tf.convert_to_tensor(UpperCamelCase__)
          if text_pair is not None:
                 if text.shape.rank > 1:
                        raise ValueError("text argument should not be multidimensional when a text pair is supplied!")
                 if text_pair.shape.rank > 1:
                        raise ValueError("text_pair should not be multidimensional!")
          if text.shape.rank == 2:
                 lowercase__:							Union[str, Any]         =     text[:, 0], text[:, 1]
          lowercase__:							List[str]         =     self.unpaired_tokenize(UpperCamelCase__)
          if text_pair is None:  # Unpaired text
                 if truncation:
                        lowercase__:							Optional[int]         =     text[:, : max_length - 2]  # Allow room for special tokens
                 lowercase__:							Tuple         =     combine_segments(
                     (text,)					, start_of_sequence_id=self.cls_token_id					, end_of_segment_id=self.sep_token_id)
          else:  # Paired text
                 lowercase__:							Union[str, Any]         =     self.unpaired_tokenize(UpperCamelCase__)
                 if truncation:
                        lowercase__:							Any         =     self.paired_trimmer.trim([text, text_pair])
                 lowercase__:							str         =     combine_segments(
                     (text, text_pair)					, start_of_sequence_id=self.cls_token_id					, end_of_segment_id=self.sep_token_id)
          if padding == "longest":
                 lowercase__:							Any         =     input_ids.bounding_shape(axis=1)
                 if pad_to_multiple_of is not None:
                        # No ceiling division in tensorflow, so we negate floordiv instead
                        lowercase__:							Any         =     pad_to_multiple_of * (-tf.math.floordiv(-pad_length					, UpperCamelCase__))
          else:
                 lowercase__:							int         =     max_length

          lowercase__:							Any         =     pad_model_inputs(UpperCamelCase__					, max_seq_length=UpperCamelCase__					, pad_value=self.pad_token_id)
          lowercase__:							Any         =     {'''input_ids''': input_ids}
          if return_attention_mask:
                 lowercase__:							Union[str, Any]         =     attention_mask
          if return_token_type_ids:
                 lowercase__:							List[str]         =     pad_model_inputs(
                     UpperCamelCase__					, max_seq_length=UpperCamelCase__					, pad_value=self.pad_token_id)
                 lowercase__:							int         =     token_type_ids
          return output





   def __lowercase							(       self)			->   Dict:




          '''simple docstring'''




          return {
              "vocab_list": self.vocab_list,
              "do_lower_case": self.do_lower_case,
              "cls_token_id": self.cls_token_id,
              "sep_token_id": self.sep_token_id,
              "pad_token_id": self.pad_token_id,
          }








import unittest
from functools import partial

import torch

from accelerate import Accelerator, debug_launcher
from accelerate.state import AcceleratorState, GradientState
from accelerate.test_utils import require_cpu, require_huggingface_suite
from accelerate.utils import GradientAccumulationPlugin

def 					SCREAMING_SNAKE_CASE  (   __UpperCamelCase			:    Any=2				,    __UpperCamelCase			:    str=True				,    __UpperCamelCase			:    Optional[Any]=False							)			->							Union[str, Any]:

				"""simple docstring"""







				A__	:						List[str]								=				Accelerator(step_scheduler_with_optimizer=__UpperCamelCase				,    split_batches=__UpperCamelCase							)
				A__	:						str								=				torch.nn.Linear(2				,    4							)
				A__	:						Any								=				torch.optim.AdamW(model.parameters()				,    lr=1.0							)
				A__	:						Tuple								=				torch.optim.lr_scheduler.OneCycleLR(__UpperCamelCase				,    max_lr=0.0_1				,    steps_per_epoch=2				,    epochs=1							)
				A__							,				A__							,				A__	:						str								=				accelerator.prepare(__UpperCamelCase				,    __UpperCamelCase				,    __UpperCamelCase							)

				# Optimizer has stepped
				scheduler.step()
				if step_scheduler_with_optimizer or (num_processes == 1):
								assert (
								    scheduler.scheduler.last_epoch == num_processes
								), F"Last Epoch ({scheduler.scheduler.last_epoch}) != Num Processes ({num_processes})"
				else:
								assert (
								    scheduler.scheduler.last_epoch != num_processes
								), F"Last Epoch ({scheduler.scheduler.last_epoch}) == Num Processes ({num_processes})"

def 					SCREAMING_SNAKE_CASE  (   __UpperCamelCase			:    Optional[Any]=2				,    __UpperCamelCase			:    Tuple=True				,    __UpperCamelCase			:    List[Any]=False							)			->							str:

				"""simple docstring"""







				A__	:						Union[str, Any]								=				Accelerator(step_scheduler_with_optimizer=__UpperCamelCase				,    split_batches=__UpperCamelCase							)
				A__	:						List[str]								=				torch.nn.Linear(2				,    4							)
				A__	:						int								=				torch.optim.AdamW(model.parameters()				,    lr=1.0							)
				A__	:						List[Any]								=				torch.optim.lr_scheduler.LambdaLR(__UpperCamelCase				,    lr_lambda=lambda __UpperCamelCase							: 1 - n / 10							)
				A__							,				A__							,				A__	:						int								=				accelerator.prepare(__UpperCamelCase				,    __UpperCamelCase				,    __UpperCamelCase							)

				# Optimizer has stepped
				A__	:						List[Any]								=				False
				scheduler.step()
				A__	:						str								=				1 - (num_processes if (step_scheduler_with_optimizer and not split_batches) else 1) / 10
				assert (
				    scheduler.get_last_lr()[0] == expected_lr
				), F"Wrong lr found at first step, expected {expected_lr}, got {scheduler.get_last_lr()[0]}"

				# Optimizer has not stepped
				A__	:						str								=				True
				scheduler.step()
				if not step_scheduler_with_optimizer:
								A__	:						Dict								=				1 - 2 / 10
				assert (
				    scheduler.get_last_lr()[0] == expected_lr
				), F"Wrong lr found at second step, expected {expected_lr}, got {scheduler.get_last_lr()[0]}"

def 					SCREAMING_SNAKE_CASE  (   __UpperCamelCase			:    int = 2							)			->							List[Any]:

				"""simple docstring"""







				from transformers import get_linear_schedule_with_warmup

				A__	:						str								=				[1, 2, 4]
				for num_steps in steps:
								A__	:						Any								=				GradientAccumulationPlugin(num_steps=__UpperCamelCase				,    adjust_scheduler=num_steps > 1							)
								A__	:						Tuple								=				Accelerator(gradient_accumulation_plugin=__UpperCamelCase							)
								A__	:						Optional[int]								=				torch.nn.Linear(2				,    4							)
								A__	:						Optional[int]								=				torch.optim.AdamW(model.parameters()				,    lr=1_0.0							)
								A__	:						Optional[int]								=				get_linear_schedule_with_warmup(optimizer=__UpperCamelCase				,    num_warmup_steps=0				,    num_training_steps=20							)

								A__							,				A__							,				A__	:						List[Any]								=				accelerator.prepare(__UpperCamelCase				,    __UpperCamelCase				,    __UpperCamelCase							)

								for i in range(10 * num_steps							):
												with accelerator.accumulate(__UpperCamelCase							):
																optimizer.step()
																scheduler.step()

												if i == (10 * num_steps - 2):
																assert (
																    scheduler.get_last_lr()[0] != 0
																), F"Wrong lr found at second-to-last step, expected non-zero, got {scheduler.get_last_lr()[0]}. num_steps: {num_steps}"
								assert (
								    scheduler.get_last_lr()[0] == 0
								), F"Wrong lr found at last step, expected 0, got {scheduler.get_last_lr()[0]}"
								GradientState._reset_state()







@require_cpu
class 			UpperCamelCase__		(     unittest.TestCase ):







	'''simple docstring'''





	def 	__snake_case					(      self		):
					debug_launcher(partial(UpperCamelCase__					,							num_processes=1		)					,							num_processes=1		)
					debug_launcher(partial(UpperCamelCase__					,							num_processes=1					,							split_batches=UpperCamelCase__		)					,							num_processes=1		)





	def 	__snake_case					(      self		):
					debug_launcher(partial(UpperCamelCase__					,							num_processes=1		)					,							num_processes=1		)
					debug_launcher(partial(UpperCamelCase__					,							num_processes=1					,							split_batches=UpperCamelCase__		)					,							num_processes=1		)





	def 	__snake_case					(      self		):
					debug_launcher(partial(UpperCamelCase__					,							num_processes=1					,							step_scheduler_with_optimizer=UpperCamelCase__		)					,							num_processes=1		)





	def 	__snake_case					(      self		):
					debug_launcher(partial(UpperCamelCase__					,							num_processes=1					,							step_scheduler_with_optimizer=UpperCamelCase__		)					,							num_processes=1		)





	def 	__snake_case					(      self		):
					AcceleratorState._reset_state(UpperCamelCase__		)
					debug_launcher(UpperCamelCase__		)
					debug_launcher(partial(UpperCamelCase__					,							num_processes=1					,							split_batches=UpperCamelCase__		)					,							num_processes=1		)





	def 	__snake_case					(      self		):
					AcceleratorState._reset_state(UpperCamelCase__		)
					debug_launcher(UpperCamelCase__		)
					debug_launcher(partial(UpperCamelCase__					,							num_processes=1					,							split_batches=UpperCamelCase__		)					,							num_processes=1		)





	def 	__snake_case					(      self		):
					AcceleratorState._reset_state(UpperCamelCase__		)
					debug_launcher(partial(UpperCamelCase__					,							step_scheduler_with_optimizer=UpperCamelCase__		)		)





	def 	__snake_case					(      self		):
					AcceleratorState._reset_state(UpperCamelCase__		)
					debug_launcher(partial(UpperCamelCase__					,							step_scheduler_with_optimizer=UpperCamelCase__		)		)





	@require_huggingface_suite
	def 	__snake_case					(      self		):
					AcceleratorState._reset_state(UpperCamelCase__		)
					debug_launcher(partial(UpperCamelCase__					,							num_processes=1		)		)
					debug_launcher(UpperCamelCase__		)