



import re
import time
from typing import Optional

import IPython.display as disp

from ..trainer_callback import TrainerCallback
from ..trainer_utils import IntervalStrategy, has_length



def 		lowerCAmelCase__    (					UpperCamelCase_				:       str	)->  List[str]:
    A__		 =					int(UpperCamelCase_	)
    A__ ,	A__ ,	A__		 =					t // 3_6_0_0, (t // 6_0) % 6_0, t % 6_0
    return f"{h}:{m:02d}:{s:02d}" if h != 0 else f"{m:02d}:{s:02d}"



def 		lowerCAmelCase__    (					UpperCamelCase_				:       Any ,       UpperCamelCase_				:       List[Any] ,       UpperCamelCase_				:       List[Any] ,       UpperCamelCase_				:       Union[str, Any] ,       UpperCamelCase_				:       Optional[Any]=3_0_0	)->  Dict:
    return f"\n    <div>\n      {prefix}\n      <progress value=\'{value}\' max=\'{total}\' style=\'width:{width}px; height:20px; vertical-align: middle;\'></progress>\n      {label}\n    </div>\n    "



def 		lowerCAmelCase__    (					UpperCamelCase_				:       Any	)->  Optional[Any]:
    A__		 =					'''<table border="1" class="dataframe">\n'''
    html_code += """  <thead>\n <tr style="text-align: left;">\n"""
    for i in items[0]:
        html_code += f"      <th>{i}</th>\n"
    html_code += "    </tr>\n  </thead>\n  <tbody>\n"
    for line in items[1:]:
        html_code += "    <tr>\n"
        for elt in line:
            A__		 =					f"{elt:.6f}" if isinstance(UpperCamelCase_ ,       UpperCamelCase_	) else str(UpperCamelCase_	)
            html_code += f"      <td>{elt}</td>\n"
        html_code += "    </tr>\n"
    html_code += "  </tbody>\n</table><p>"
    return html_code



class     _UpperCAmelCase			:
     UpperCamelCase__					    =				5
     UpperCamelCase__					    =				0.2




     def __init__(				self ,						a__ ,						a__ = None ,						a__ = True ,						a__ = None ,						a__ = 3_0_0 ,						):
         A__		 =					total
         A__		 =					'''''' if prefix is None else prefix
         A__		 =					leave
         A__		 =					parent
         A__		 =					width
         A__		 =					None
         A__		 =					None
         A__		 =					None




     def 		snake_case_		(				self ,						a__ ,						a__ = False ,						a__ = None):
         A__		 =					value
         if comment is not None:
             A__		 =					comment
         if self.last_value is None:
             A__		 =					A__		 =					time.time()
             A__		 =					A__		 =					value
             A__		 =					A__		 =					None
             A__		 =					self.warmup
             A__		 =					1
             self.update_bar(a__)
         elif value <= self.last_value and not force_update:
             return
         elif force_update or self.first_calls > 0 or value >= min(self.last_value + self.wait_for ,						self.total):
             if self.first_calls > 0:
                 self.first_calls -= 1
             A__		 =					time.time()
             A__		 =					current_time - self.start_time
             # We could have value = self.start_value if the update is called twixe with the same start value.
             if value > self.start_value:
                 A__		 =					self.elapsed_time / (value - self.start_value)
             else:
                 A__		 =					None
             if value >= self.total:
                 A__		 =					self.total
                 A__		 =					None
                 if not self.leave:
                     self.close()
             elif self.average_time_per_item is not None:
                 A__		 =					self.average_time_per_item * (self.total - value)
             self.update_bar(a__)
             A__		 =					value
             A__		 =					current_time
             if self.average_time_per_item is None:
                 A__		 =					1
             else:
                 A__		 =					max(int(self.update_every / self.average_time_per_item) ,						1)




     def 		snake_case_		(				self ,						a__ ,						a__=None):
         A__		 =					''' ''' * (len(str(self.total)) - len(str(a__))) + str(a__)
         if self.elapsed_time is None:
             A__		 =					F"[{spaced_value}/{self.total} : < :"
         elif self.predicted_remaining is None:
             A__		 =					F"[{spaced_value}/{self.total} {format_time(self.elapsed_time)}"
         else:
             A__		 =					(
                 F"[{spaced_value}/{self.total} {format_time(self.elapsed_time)} <"
                 F" {format_time(self.predicted_remaining)}"
             )
             self.label += F", {1/self.average_time_per_item:.2f} it/s"
         self.label += "]" if self.comment is None or len(self.comment) == 0 else F", {self.comment}]"
         self.display()




     def 		snake_case_		(				self):
         A__		 =					html_progress_bar(self.value ,						self.total ,						self.prefix ,						self.label ,						self.width)
         if self.parent is not None:
             # If this is a child bar, the parent will take care of the display.
             self.parent.display()
             return
         if self.output is None:
             A__		 =					disp.display(disp.HTML(self.html_code) ,						display_id=a__)
         else:
             self.output.update(disp.HTML(self.html_code))




     def 		snake_case_		(				self):
         if self.parent is None and self.output is not None:
             self.output.update(disp.HTML(''''''))







class     _UpperCAmelCase			(       A__     ):




     def __init__(				self ,						a__ ,						a__=None):
         super().__init__(a__)
         A__		 =					None if column_names is None else [column_names]
         A__		 =					None




     def 		snake_case_		(				self):
         A__		 =					html_progress_bar(self.value ,						self.total ,						self.prefix ,						self.label ,						self.width)
         if self.inner_table is not None:
             self.html_code += text_to_html_table(self.inner_table)
         if self.child_bar is not None:
             self.html_code += self.child_bar.html_code
         if self.output is None:
             A__		 =					disp.display(disp.HTML(self.html_code) ,						display_id=a__)
         else:
             self.output.update(disp.HTML(self.html_code))




     def 		snake_case_		(				self ,						a__):
         if self.inner_table is None:
             A__		 =					[list(values.keys()), list(values.values())]
         else:
             A__		 =					self.inner_table[0]
             if len(self.inner_table) == 1:
                 # We give a chance to update the column names at the first iteration
                 for key in values.keys():
                     if key not in columns:
                         columns.append(a__)
                 A__		 =					columns
             self.inner_table.append([values[c] for c in columns])




     def 		snake_case_		(				self ,						a__ ,						a__=None ,						a__=3_0_0):
         A__		 =					NotebookProgressBar(a__ ,						prefix=a__ ,						parent=self ,						width=a__)
         return self.child_bar




     def 		snake_case_		(				self):
         A__		 =					None
         self.display()







class     _UpperCAmelCase			(       A__     ):




     def __init__(				self):
         A__		 =					None
         A__		 =					None
         A__		 =					False




     def 		snake_case_		(				self ,						a__ ,						a__ ,						a__ ,						**a__):
         A__		 =					'''Epoch''' if args.evaluation_strategy == IntervalStrategy.EPOCH else '''Step'''
         A__		 =					0
         A__		 =					0
         A__		 =					[self.first_column] + ['''Training Loss''']
         if args.evaluation_strategy != IntervalStrategy.NO:
             column_names.append('''Validation Loss''')
         A__		 =					NotebookTrainingTracker(state.max_steps ,						a__)




     def 		snake_case_		(				self ,						a__ ,						a__ ,						a__ ,						**a__):
         A__		 =					int(state.epoch) if int(state.epoch) == state.epoch else F"{state.epoch:.2f}"
         self.training_tracker.update(
             state.global_step + 1 ,						comment=F"Epoch {epoch}/{state.num_train_epochs}" ,						force_update=self._force_next_update ,						)
         A__		 =					False




     def 		snake_case_		(				self ,						a__ ,						a__ ,						a__ ,						a__=None ,						**a__):
         if not has_length(a__):
             return
         if self.prediction_bar is None:
             if self.training_tracker is not None:
                 A__		 =					self.training_tracker.add_child(len(a__))
             else:
                 A__		 =					NotebookProgressBar(len(a__))
             self.prediction_bar.update(1)
         else:
             self.prediction_bar.update(self.prediction_bar.value + 1)




     def 		snake_case_		(				self ,						a__ ,						a__ ,						a__ ,						**a__):
         if self.prediction_bar is not None:
             self.prediction_bar.close()
         A__		 =					None




     def 		snake_case_		(				self ,						a__ ,						a__ ,						a__ ,						a__=None ,						**a__):
         # Only for when there is no evaluation
         if args.evaluation_strategy == IntervalStrategy.NO and "loss" in logs:
             A__		 =					{'''Training Loss''': logs['''loss''']}
             # First column is necessarily Step sine we're not in epoch eval strategy
             A__		 =					state.global_step
             self.training_tracker.write_line(a__)




     def 		snake_case_		(				self ,						a__ ,						a__ ,						a__ ,						a__=None ,						**a__):
         if self.training_tracker is not None:
             A__		 =					{'''Training Loss''': '''No log''', '''Validation Loss''': '''No log'''}
             for log in reversed(state.log_history):
                 if "loss" in log:
                     A__		 =					log['''loss''']
                     break

             if self.first_column == "Epoch":
                 A__		 =					int(state.epoch)
             else:
                 A__		 =					state.global_step
             A__		 =					'''eval'''
             for k in metrics:
                 if k.endswith('''_loss'''):
                     A__		 =					re.sub(R'''\_loss$''' ,						'''''' ,						a__)
             A__		 =					metrics.pop('''total_flos''' ,						a__)
             A__		 =					metrics.pop('''epoch''' ,						a__)
             A__		 =					metrics.pop(F"{metric_key_prefix}_runtime" ,						a__)
             A__		 =					metrics.pop(F"{metric_key_prefix}_samples_per_second" ,						a__)
             A__		 =					metrics.pop(F"{metric_key_prefix}_steps_per_second" ,						a__)
             A__		 =					metrics.pop(F"{metric_key_prefix}_jit_compilation_time" ,						a__)
             for k, v in metrics.items():
                 if k == F"{metric_key_prefix}_loss":
                     A__		 =					v
                 else:
                     A__		 =					k.split('''_''')
                     A__		 =					''' '''.join([part.capitalize() for part in splits[1:]])
                     A__		 =					v
             self.training_tracker.write_line(a__)
             self.training_tracker.remove_child()
             A__		 =					None
             # Evaluation takes a long time so we should force the next update.
             A__		 =					True




     def 		snake_case_		(				self ,						a__ ,						a__ ,						a__ ,						**a__):
         self.training_tracker.update(
             state.global_step ,						comment=F"Epoch {int(state.epoch)}/{state.num_train_epochs}" ,						force_update=a__)
         A__		 =					None





from __future__ import annotations

import unittest

from transformers import is_tf_available
from transformers.testing_utils import require_sentencepiece, require_tf, require_tokenizers, slow

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
   import numpy as np
   import tensorflow as tf

   from transformers import (
       TF_FLAUBERT_PRETRAINED_MODEL_ARCHIVE_LIST,
       FlaubertConfig,
       TFFlaubertForMultipleChoice,
       TFFlaubertForQuestionAnsweringSimple,
       TFFlaubertForSequenceClassification,
       TFFlaubertForTokenClassification,
       TFFlaubertModel,
       TFFlaubertWithLMHeadModel,
   )



class     _UpperCAmelCase			:




     def __init__(				self ,						a__ ,						):
         A__		 =					parent
         A__		 =					1_3
         A__		 =					7
         A__		 =					True
         A__		 =					True
         A__		 =					True
         A__		 =					True
         A__		 =					True
         A__		 =					False
         A__		 =					False
         A__		 =					False
         A__		 =					2
         A__		 =					9_9
         A__		 =					0
         A__		 =					3_2
         A__		 =					2
         A__		 =					4
         A__		 =					0.1
         A__		 =					0.1
         A__		 =					5_1_2
         A__		 =					1_6
         A__		 =					2
         A__		 =					0.0_2
         A__		 =					3
         A__		 =					4
         A__		 =					'''last'''
         A__		 =					True
         A__		 =					None
         A__		 =					0




     def 		snake_case_		(				self):
         A__		 =					ids_tensor([self.batch_size, self.seq_length] ,						self.vocab_size)
         A__		 =					random_attention_mask([self.batch_size, self.seq_length] ,						dtype=tf.floataa)

         A__		 =					None
         if self.use_input_lengths:
             A__		 =					(
                 ids_tensor([self.batch_size] ,						vocab_size=2) + self.seq_length - 2
             )  # small variation of seq_length

         A__		 =					None
         if self.use_token_type_ids:
             A__		 =					ids_tensor([self.batch_size, self.seq_length] ,						self.n_langs)

         A__		 =					None
         A__		 =					None
         A__		 =					None
         if self.use_labels:
             A__		 =					ids_tensor([self.batch_size] ,						self.type_sequence_label_size)
             A__		 =					ids_tensor([self.batch_size, self.seq_length] ,						self.num_labels)
             A__		 =					ids_tensor([self.batch_size] ,						2 ,						dtype=tf.floataa)
             A__		 =					ids_tensor([self.batch_size] ,						self.num_choices)

         A__		 =					FlaubertConfig(
             vocab_size=self.vocab_size ,						n_special=self.n_special ,						emb_dim=self.hidden_size ,						n_layers=self.num_hidden_layers ,						n_heads=self.num_attention_heads ,						dropout=self.hidden_dropout_prob ,						attention_dropout=self.attention_probs_dropout_prob ,						gelu_activation=self.gelu_activation ,						sinusoidal_embeddings=self.sinusoidal_embeddings ,						asm=self.asm ,						causal=self.causal ,						n_langs=self.n_langs ,						max_position_embeddings=self.max_position_embeddings ,						initializer_range=self.initializer_range ,						summary_type=self.summary_type ,						use_proj=self.use_proj ,						bos_token_id=self.bos_token_id ,						)

         return (
             config,
             input_ids,
             token_type_ids,
             input_lengths,
             sequence_labels,
             token_labels,
             is_impossible_labels,
             choice_labels,
             input_mask,
         )




     def 		snake_case_		(				self ,						a__ ,						a__ ,						a__ ,						a__ ,						a__ ,						a__ ,						a__ ,						a__ ,						a__ ,						):
         A__		 =					TFFlaubertModel(config=a__)
         A__		 =					{'''input_ids''': input_ids, '''lengths''': input_lengths, '''langs''': token_type_ids}
         A__		 =					model(a__)

         A__		 =					[input_ids, input_mask]
         A__		 =					model(a__)
         self.parent.assertEqual(result.last_hidden_state.shape ,						(self.batch_size, self.seq_length, self.hidden_size))




     def 		snake_case_		(				self ,						a__ ,						a__ ,						a__ ,						a__ ,						a__ ,						a__ ,						a__ ,						a__ ,						a__ ,						):
         A__		 =					TFFlaubertWithLMHeadModel(a__)

         A__		 =					{'''input_ids''': input_ids, '''lengths''': input_lengths, '''langs''': token_type_ids}
         A__		 =					model(a__)

         self.parent.assertEqual(result.logits.shape ,						(self.batch_size, self.seq_length, self.vocab_size))




     def 		snake_case_		(				self ,						a__ ,						a__ ,						a__ ,						a__ ,						a__ ,						a__ ,						a__ ,						a__ ,						a__ ,						):
         A__		 =					TFFlaubertForQuestionAnsweringSimple(a__)

         A__		 =					{'''input_ids''': input_ids, '''lengths''': input_lengths}

         A__		 =					model(a__)

         self.parent.assertEqual(result.start_logits.shape ,						(self.batch_size, self.seq_length))
         self.parent.assertEqual(result.end_logits.shape ,						(self.batch_size, self.seq_length))




     def 		snake_case_		(				self ,						a__ ,						a__ ,						a__ ,						a__ ,						a__ ,						a__ ,						a__ ,						a__ ,						a__ ,						):
         A__		 =					TFFlaubertForSequenceClassification(a__)

         A__		 =					{'''input_ids''': input_ids, '''lengths''': input_lengths}

         A__		 =					model(a__)

         self.parent.assertEqual(result.logits.shape ,						(self.batch_size, self.type_sequence_label_size))




     def 		snake_case_		(				self ,						a__ ,						a__ ,						a__ ,						a__ ,						a__ ,						a__ ,						a__ ,						a__ ,						a__ ,						):
         A__		 =					self.num_labels
         A__		 =					TFFlaubertForTokenClassification(config=a__)
         A__		 =					{'''input_ids''': input_ids, '''attention_mask''': input_mask, '''token_type_ids''': token_type_ids}
         A__		 =					model(a__)
         self.parent.assertEqual(result.logits.shape ,						(self.batch_size, self.seq_length, self.num_labels))




     def 		snake_case_		(				self ,						a__ ,						a__ ,						a__ ,						a__ ,						a__ ,						a__ ,						a__ ,						a__ ,						a__ ,						):
         A__		 =					self.num_choices
         A__		 =					TFFlaubertForMultipleChoice(config=a__)
         A__		 =					tf.tile(tf.expand_dims(a__ ,						1) ,						(1, self.num_choices, 1))
         A__		 =					tf.tile(tf.expand_dims(a__ ,						1) ,						(1, self.num_choices, 1))
         A__		 =					tf.tile(tf.expand_dims(a__ ,						1) ,						(1, self.num_choices, 1))
         A__		 =					{
             '''input_ids''': multiple_choice_inputs_ids,
             '''attention_mask''': multiple_choice_input_mask,
             '''token_type_ids''': multiple_choice_token_type_ids,
         }
         A__		 =					model(a__)
         self.parent.assertEqual(result.logits.shape ,						(self.batch_size, self.num_choices))




     def 		snake_case_		(				self):
         A__		 =					self.prepare_config_and_inputs()
         (
             (
             A__
         ) ,	(
             A__
         ) ,	(
             A__
         ) ,	(
             A__
         ) ,	(
             A__
         ) ,	(
             A__
         ) ,	(
             A__
         ) ,	(
             A__
         ) ,	(
             A__
         ) ,	
         )		 =					config_and_inputs
         A__		 =					{
             '''input_ids''': input_ids,
             '''token_type_ids''': token_type_ids,
             '''langs''': token_type_ids,
             '''lengths''': input_lengths,
         }
         return config, inputs_dict







@require_tf
class     _UpperCAmelCase			(       A__			,      A__			,      unittest.TestCase     ):
     UpperCamelCase__		    =				(
         (
             TFFlaubertModel,
             TFFlaubertWithLMHeadModel,
             TFFlaubertForSequenceClassification,
             TFFlaubertForQuestionAnsweringSimple,
             TFFlaubertForTokenClassification,
             TFFlaubertForMultipleChoice,
         )
         if is_tf_available()
         else ()
     )
     UpperCamelCase__		    =				(
         (TFFlaubertWithLMHeadModel,) if is_tf_available() else ()
     )  # TODO (PVP): Check other models whether language generation is also applicable
     UpperCamelCase__		    =				(
         {
             '''feature-extraction''': TFFlaubertModel,
             '''fill-mask''': TFFlaubertWithLMHeadModel,
             '''question-answering''': TFFlaubertForQuestionAnsweringSimple,
             '''text-classification''': TFFlaubertForSequenceClassification,
             '''token-classification''': TFFlaubertForTokenClassification,
             '''zero-shot''': TFFlaubertForSequenceClassification,
         }
         if is_tf_available()
         else {}
     )
     UpperCamelCase__		    =				False
     UpperCamelCase__		    =				False




     def 		snake_case_		(				self ,						a__ ,						a__ ,						a__ ,						a__ ,						a__):
         if (
             pipeline_test_casse_name == "QAPipelineTests"
             and tokenizer_name is not None
             and not tokenizer_name.endswith('''Fast''')
         ):
             # `QAPipelineTests` fails for a few models when the slower tokenizer are used.
             # (The slower tokenizers were never used for pipeline tests before the pipeline testing rework)
             # TODO: check (and possibly fix) the `QAPipelineTests` with slower tokenizer
             return True

         return False




     def 		snake_case_		(				self):
         A__		 =					TFFlaubertModelTester(self)
         A__		 =					ConfigTester(self ,						config_class=a__ ,						emb_dim=3_7)




     def 		snake_case_		(				self):
         self.config_tester.run_common_tests()




     def 		snake_case_		(				self):
         A__		 =					self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_flaubert_model(*a__)




     def 		snake_case_		(				self):
         A__		 =					self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_flaubert_lm_head(*a__)




     def 		snake_case_		(				self):
         A__		 =					self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_flaubert_qa(*a__)




     def 		snake_case_		(				self):
         A__		 =					self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_flaubert_sequence_classif(*a__)




     def 		snake_case_		(				self):
         A__		 =					self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_flaubert_for_token_classification(*a__)




     def 		snake_case_		(				self):
         A__		 =					self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_flaubert_for_multiple_choice(*a__)




     @slow
     def 		snake_case_		(				self):
         for model_name in TF_FLAUBERT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
             A__		 =					TFFlaubertModel.from_pretrained(a__)
             self.assertIsNotNone(a__)







@require_tf
@require_sentencepiece
@require_tokenizers
class     _UpperCAmelCase			(       unittest.TestCase     ):




     @slow
     def 		snake_case_		(				self):
         A__		 =					TFFlaubertModel.from_pretrained('''jplu/tf-flaubert-small-cased''')

         A__		 =					tf.convert_to_tensor(
             [[0, 1_5_8, 7_3_5, 2_5_9_2, 1_4_2_4, 6_7_2_7, 8_2, 1]] ,						dtype=tf.intaa ,						)  # "J'aime flaubert !"

         A__		 =					model(a__)[0]
         A__		 =					tf.TensorShape((1, 8, 5_1_2))
         self.assertEqual(output.shape ,						a__)
         # compare the actual values for a slice.
         A__		 =					tf.convert_to_tensor(
             [
                 [
                     [-1.8_7_6_8_7_7_3, -1.5_6_6_5_5_5, 0.2_7_0_7_2_4_1_8],
                     [-1.6_9_2_0_0_3_8, -0.5_8_7_3_5_0_5, 1.9_3_2_9_5_9_9],
                     [-2.9_5_6_3_9_8_5, -1.6_9_9_3_8_3_5, 1.7_9_7_2_0_5_2],
                 ]
             ] ,						dtype=tf.floataa ,						)

         self.assertTrue(np.allclose(output[:, :3, :3].numpy() ,						expected_slice.numpy() ,						atol=1e-4))
