"""simple docstring"""
import os
from argparse import ArgumentParser, Namespace

from ..data import SingleSentenceClassificationProcessor as Processor
from ..pipelines import TextClassificationPipeline
from ..utils import is_tf_available, is_torch_available, logging
from . import BaseTransformersCLICommand


if not is_tf_available() and not is_torch_available():
 raise RuntimeError('At least one of PyTorch or TensorFlow 2.0+ should be installed to use CLI training')

# TF training parameters
lowercase__ :    Optional[int] 	=     False
lowercase__ :    int 	=     False







def 						__a       (     _SCREAMING_SNAKE_CASE       )     ->List[str]:
  return TrainCommand(_SCREAMING_SNAKE_CASE       )




class        __snake_case	(					__lowerCAmelCase    ):



      @staticmethod
      def   lowerCamelCase_       (						lowercase)     ->      Dict:





        '''simple docstring'''





        a__:					Dict      							=     parser.add_parser('train'   ,      help='CLI tool to train a model on a task.')

        train_parser.add_argument(
            '--train_data'   ,      type=lowercase   ,      required=lowercase   ,      help='path to train (and optionally evaluation) dataset as a csv with tab separated labels and sentences.'   ,      )
        train_parser.add_argument(
            '--column_label'   ,      type=lowercase   ,      default=0   ,      help='Column of the dataset csv file with example labels.')
        train_parser.add_argument(
            '--column_text'   ,      type=lowercase   ,      default=1   ,      help='Column of the dataset csv file with example texts.')
        train_parser.add_argument(
            '--column_id'   ,      type=lowercase   ,      default=2   ,      help='Column of the dataset csv file with example ids.')
        train_parser.add_argument(
            '--skip_first_row'   ,      action='store_true'   ,      help='Skip the first row of the csv file (headers).')

        train_parser.add_argument('--validation_data'   ,      type=lowercase   ,      default=''   ,      help='path to validation dataset.')
        train_parser.add_argument(
            '--validation_split'   ,      type=lowercase   ,      default=0.1   ,      help='if validation dataset is not provided, fraction of train dataset to use as validation dataset.'   ,      )

        train_parser.add_argument('--output'   ,      type=lowercase   ,      default='./'   ,      help='path to saved the trained model.')

        train_parser.add_argument(
            '--task'   ,      type=lowercase   ,      default='text_classification'   ,      help='Task to train the model on.')
        train_parser.add_argument(
            '--model'   ,      type=lowercase   ,      default='bert-base-uncased'   ,      help='Model\'s name or path to stored model.')
        train_parser.add_argument('--train_batch_size'   ,      type=lowercase   ,      default=32   ,      help='Batch size for training.')
        train_parser.add_argument('--valid_batch_size'   ,      type=lowercase   ,      default=64   ,      help='Batch size for validation.')
        train_parser.add_argument('--learning_rate'   ,      type=lowercase   ,      default=3e-5   ,      help='Learning rate.')
        train_parser.add_argument('--adam_epsilon'   ,      type=lowercase   ,      default=1e-08   ,      help='Epsilon for Adam optimizer.')
        train_parser.set_defaults(func=lowercase)



      def __init__(						self   ,      lowercase)     ->      int:





        '''simple docstring'''





        a__:					List[Any]      							=     logging.get_logger('transformers-cli/training')

        a__:					Tuple      							=     'tf' if is_tf_available() else 'torch'

        os.makedirs(args.output   ,      exist_ok=lowercase)
        a__:					Optional[Any]      							=     args.output

        a__:					Union[str, Any]      							=     args.column_label
        a__:					Tuple      							=     args.column_text
        a__:					int      							=     args.column_id

        self.logger.info(f'Loading {args.task} pipeline for {args.model}')
        if args.task == "text_classification":
          a__:					Tuple      							=     TextClassificationPipeline.from_pretrained(args.model)
        elif args.task == "token_classification":
          raise NotImplementedError
        elif args.task == "question_answering":
          raise NotImplementedError

        self.logger.info(f'Loading dataset from {args.train_data}')
        a__:					Union[str, Any]      							=     Processor.create_from_csv(
            args.train_data   ,      column_label=args.column_label   ,      column_text=args.column_text   ,      column_id=args.column_id   ,      skip_first_row=args.skip_first_row   ,      )
        a__:					List[str]      							=     None
        if args.validation_data:
          self.logger.info(f'Loading validation dataset from {args.validation_data}')
          a__:					Optional[Any]      							=     Processor.create_from_csv(
              args.validation_data   ,      column_label=args.column_label   ,      column_text=args.column_text   ,      column_id=args.column_id   ,      skip_first_row=args.skip_first_row   ,      )

        a__:					str      							=     args.validation_split
        a__:					Optional[int]      							=     args.train_batch_size
        a__:					Tuple      							=     args.valid_batch_size
        a__:					str      							=     args.learning_rate
        a__:					Optional[Any]      							=     args.adam_epsilon



      def   lowerCamelCase_       (						self)     ->      str:





        '''simple docstring'''





        if self.framework == "tf":
          return self.run_tf()
        return self.run_torch()



      def   lowerCamelCase_       (						self)     ->      Any:





        '''simple docstring'''





        raise NotImplementedError



      def   lowerCamelCase_       (						self)     ->      Union[str, Any]:





        '''simple docstring'''





        self.pipeline.fit(
            self.train_dataset   ,      validation_data=self.valid_dataset   ,      validation_split=self.validation_split   ,      learning_rate=self.learning_rate   ,      adam_epsilon=self.adam_epsilon   ,      train_batch_size=self.train_batch_size   ,      valid_batch_size=self.valid_batch_size   ,      )

        # Save trained pipeline
        self.pipeline.save_pretrained(self.output)



"""simple docstring"""
from __future__ import annotations

import unittest

import numpy as np

from transformers import BlipTextConfig
from transformers.testing_utils import require_tf, slow
from transformers.utils import is_tf_available

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


if is_tf_available():
 import tensorflow as tf

 from transformers import TFBlipTextModel
 from transformers.models.blip.modeling_tf_blip import TF_BLIP_PRETRAINED_MODEL_ARCHIVE_LIST




class        __snake_case	:



      def __init__(						self   ,      lowercase   ,      lowercase=12   ,      lowercase=7   ,      lowercase=True   ,      lowercase=True   ,      lowercase=True   ,      lowercase=99   ,      lowercase=32   ,      lowercase=32   ,      lowercase=2   ,      lowercase=4   ,      lowercase=37   ,      lowercase=0.1   ,      lowercase=0.1   ,      lowercase=5_12   ,      lowercase=0.02   ,      lowercase=0   ,      lowercase=None   ,      )     ->      Optional[int]:





        '''simple docstring'''





        a__:					List[Any]      							=     parent
        a__:					Any      							=     batch_size
        a__:					int      							=     seq_length
        a__:					List[str]      							=     is_training
        a__:					Any      							=     use_input_mask
        a__:					Optional[Any]      							=     use_labels
        a__:					List[Any]      							=     vocab_size
        a__:					Optional[Any]      							=     hidden_size
        a__:					Any      							=     projection_dim
        a__:					List[str]      							=     num_hidden_layers
        a__:					Dict      							=     num_attention_heads
        a__:					int      							=     intermediate_size
        a__:					Tuple      							=     dropout
        a__:					Union[str, Any]      							=     attention_dropout
        a__:					str      							=     max_position_embeddings
        a__:					List[str]      							=     initializer_range
        a__:					Optional[int]      							=     scope
        a__:					List[Any]      							=     bos_token_id



      def   lowerCamelCase_       (						self)     ->      Dict:





        '''simple docstring'''





        a__:					str      							=     ids_tensor([self.batch_size, self.seq_length]   ,      self.vocab_size)

        a__:					str      							=     None
        if self.use_input_mask:
          a__:					List[str]      							=     random_attention_mask([self.batch_size, self.seq_length])

        if input_mask is not None:
          a__:					str      							=     input_mask.numpy()
          a__							,   a__:					Any      							=     input_mask.shape
          a__:					Tuple      							=     np.random.randint(1   ,      seq_length - 1   ,      size=(batch_size,))
          for batch_idx, start_index in enumerate(lowercase):
            a__:					Tuple      							=     1
            a__:					Any      							=     0

        a__:					Any      							=     self.get_config()

        return config, input_ids, tf.convert_to_tensor(lowercase)



      def   lowerCamelCase_       (						self)     ->      Union[str, Any]:





        '''simple docstring'''





        return BlipTextConfig(
            vocab_size=self.vocab_size   ,      hidden_size=self.hidden_size   ,      projection_dim=self.projection_dim   ,      num_hidden_layers=self.num_hidden_layers   ,      num_attention_heads=self.num_attention_heads   ,      intermediate_size=self.intermediate_size   ,      dropout=self.dropout   ,      attention_dropout=self.attention_dropout   ,      max_position_embeddings=self.max_position_embeddings   ,      initializer_range=self.initializer_range   ,      bos_token_id=self.bos_token_id   ,      )



      def   lowerCamelCase_       (						self   ,      lowercase   ,      lowercase   ,      lowercase)     ->      List[Any]:





        '''simple docstring'''





        a__:					Any      							=     TFBlipTextModel(config=lowercase)
        a__:					Union[str, Any]      							=     model(lowercase   ,      attention_mask=lowercase   ,      training=lowercase)
        a__:					List[Any]      							=     model(lowercase   ,      training=lowercase)
        self.parent.assertEqual(result.last_hidden_state.shape   ,      (self.batch_size, self.seq_length, self.hidden_size))
        self.parent.assertEqual(result.pooler_output.shape   ,      (self.batch_size, self.hidden_size))



      def   lowerCamelCase_       (						self)     ->      int:





        '''simple docstring'''





        a__:					Tuple      							=     self.prepare_config_and_inputs()
        a__							,   a__							,   a__:					Optional[Any]      							=     config_and_inputs
        a__:					Dict      							=     {'input_ids': input_ids, 'attention_mask': input_mask}
        return config, inputs_dict







@require_tf
class        __snake_case	(					__lowerCAmelCase	,				unittest.TestCase    ):
      a__				=		(TFBlipTextModel,) if is_tf_available() else ()
      a__				=		False
      a__				=		False
      a__				=		False



      def   lowerCamelCase_       (						self)     ->      Any:





        '''simple docstring'''





        a__:					List[Any]      							=     BlipTextModelTester(self)
        a__:					int      							=     ConfigTester(self   ,      config_class=lowercase   ,      hidden_size=37)



      def   lowerCamelCase_       (						self)     ->      Optional[Any]:





        '''simple docstring'''





        self.config_tester.run_common_tests()



      def   lowerCamelCase_       (						self)     ->      Union[str, Any]:





        '''simple docstring'''





        a__:					List[str]      							=     self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_model(*lowercase)



      def   lowerCamelCase_       (						self)     ->      Dict:





        '''simple docstring'''





        pass



      def   lowerCamelCase_       (						self)     ->      List[str]:





        '''simple docstring'''





        pass



      @unittest.skip(reason='Blip does not use inputs_embeds')
      def   lowerCamelCase_       (						self)     ->      List[str]:





        '''simple docstring'''





        pass



      @unittest.skip(reason='BlipTextModel has no base class and is not available in MODEL_MAPPING')
      def   lowerCamelCase_       (						self)     ->      Any:





        '''simple docstring'''





        pass



      @unittest.skip(reason='BlipTextModel has no base class and is not available in MODEL_MAPPING')
      def   lowerCamelCase_       (						self)     ->      Tuple:





        '''simple docstring'''





        pass



      @slow
      def   lowerCamelCase_       (						self)     ->      List[Any]:





        '''simple docstring'''





        for model_name in TF_BLIP_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
          a__:					Tuple      							=     TFBlipTextModel.from_pretrained(lowercase)
          self.assertIsNotNone(lowercase)



      def   lowerCamelCase_       (						self   ,      lowercase=True)     ->      str:





        '''simple docstring'''





        super().test_pt_tf_model_equivalence(allow_missing_keys=lowercase)


