





'''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  UpperCAmelCase     :




      def __init__(			self			: List[str]			,			__snake_case			: str			,			__snake_case			: Dict=12			,			__snake_case			: Dict=7			,			__snake_case			: str=True			,			__snake_case			: List[Any]=True			,			__snake_case			: Dict=True			,			__snake_case			: Optional[int]=99			,			__snake_case			: Dict=32			,			__snake_case			: Optional[Any]=32			,			__snake_case			: Union[str, Any]=2			,			__snake_case			: List[str]=4			,			__snake_case			: Optional[int]=37			,			__snake_case			: Union[str, Any]=0.1			,			__snake_case			: Optional[int]=0.1			,			__snake_case			: int=5_12			,			__snake_case			: List[Any]=0.02			,			__snake_case			: Any=0			,			__snake_case			: List[Any]=None			,			)				->      int:
            _lowerCAmelCase							     =							parent
            _lowerCAmelCase							     =							batch_size
            _lowerCAmelCase							     =							seq_length
            _lowerCAmelCase							     =							is_training
            _lowerCAmelCase							     =							use_input_mask
            _lowerCAmelCase							     =							use_labels
            _lowerCAmelCase							     =							vocab_size
            _lowerCAmelCase							     =							hidden_size
            _lowerCAmelCase							     =							projection_dim
            _lowerCAmelCase							     =							num_hidden_layers
            _lowerCAmelCase							     =							num_attention_heads
            _lowerCAmelCase							     =							intermediate_size
            _lowerCAmelCase							     =							dropout
            _lowerCAmelCase							     =							attention_dropout
            _lowerCAmelCase							     =							max_position_embeddings
            _lowerCAmelCase							     =							initializer_range
            _lowerCAmelCase							     =							scope
            _lowerCAmelCase							     =							bos_token_id




      def 				lowercase__    (			self			: List[Any]  )				->      List[str]:
            _lowerCAmelCase							     =							ids_tensor([self.batch_size, self.seq_length]			,			self.vocab_size  )

            _lowerCAmelCase							     =							None
            if self.use_input_mask:
                  _lowerCAmelCase							     =							random_attention_mask([self.batch_size, self.seq_length]  )

            if input_mask is not None:
                  _lowerCAmelCase							     =							input_mask.numpy()
                  _lowerCAmelCase     ,							_lowerCAmelCase							     =							input_mask.shape
                  _lowerCAmelCase							     =							np.random.randint(1			,			seq_length - 1			,			size=(batch_size,)  )
                  for batch_idx, start_index in enumerate(__snake_case  ):
                        _lowerCAmelCase							     =							1
                        _lowerCAmelCase							     =							0

            _lowerCAmelCase							     =							self.get_config()

            return config, input_ids, tf.convert_to_tensor(__snake_case  )




      def 				lowercase__    (			self			: List[str]  )				->      Optional[int]:
            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 				lowercase__    (			self			: int			,			__snake_case			: Any			,			__snake_case			: List[str]			,			__snake_case			: List[Any]  )				->      Optional[int]:
            _lowerCAmelCase							     =							TFBlipTextModel(config=__snake_case  )
            _lowerCAmelCase							     =							model(__snake_case			,			attention_mask=__snake_case			,			training=__snake_case  )
            _lowerCAmelCase							     =							model(__snake_case			,			training=__snake_case  )
            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 				lowercase__    (			self			: Optional[int]  )				->      Any:
            _lowerCAmelCase							     =							self.prepare_config_and_inputs()
            _lowerCAmelCase     ,							_lowerCAmelCase     ,							_lowerCAmelCase							     =							config_and_inputs
            _lowerCAmelCase							     =							{"""input_ids""": input_ids, """attention_mask""": input_mask}
            return config, inputs_dict




@require_tf
class  UpperCAmelCase     (			snake_case_							,		unittest.TestCase   ):
      _lowercase:			str						     =   (TFBlipTextModel,) if is_tf_available() else ()
      _lowercase:			Dict						     =   False
      _lowercase:			Dict						     =   False
      _lowercase:			Dict						     =   False




      def 				lowercase__    (			self			: List[Any]  )				->      Optional[int]:
            _lowerCAmelCase							     =							BlipTextModelTester(self  )
            _lowerCAmelCase							     =							ConfigTester(self			,			config_class=__snake_case			,			hidden_size=37  )




      def 				lowercase__    (			self			: Optional[Any]  )				->      Optional[Any]:
            self.config_tester.run_common_tests()




      def 				lowercase__    (			self			: int  )				->      List[str]:
            _lowerCAmelCase							     =							self.model_tester.prepare_config_and_inputs()
            self.model_tester.create_and_check_model(*__snake_case  )




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




      def 				lowercase__    (			self			: Optional[Any]  )				->      Tuple:
            pass




      @unittest.skip(reason="""Blip does not use inputs_embeds"""  )
      def 				lowercase__    (			self			: List[Any]  )				->      List[Any]:
            pass




      @unittest.skip(reason="""BlipTextModel has no base class and is not available in MODEL_MAPPING"""  )
      def 				lowercase__    (			self			: List[Any]  )				->      List[str]:
            pass




      @unittest.skip(reason="""BlipTextModel has no base class and is not available in MODEL_MAPPING"""  )
      def 				lowercase__    (			self			: List[str]  )				->      int:
            pass




      @slow
      def 				lowercase__    (			self			: List[str]  )				->      List[str]:
            for model_name in TF_BLIP_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
                  _lowerCAmelCase							     =							TFBlipTextModel.from_pretrained(__snake_case  )
                  self.assertIsNotNone(__snake_case  )




      def 				lowercase__    (			self			: List[str]			,			__snake_case			: List[str]=True  )				->      Any:
            super().test_pt_tf_model_equivalence(allow_missing_keys=__snake_case  )









'''simple docstring'''



import math
from enum import Enum
from typing import Optional, Union

from torch.optim import Optimizer
from torch.optim.lr_scheduler import LambdaLR

from .utils import logging


UpperCAmelCase :       str			   =     logging.get_logger(__name__)







class 						lowerCAmelCase__   (      a					):



 """simple docstring"""



 lowerCAmelCase__  					=      "linear"
 lowerCAmelCase__  					=      "cosine"
 lowerCAmelCase__  					=      "cosine_with_restarts"
 lowerCAmelCase__  					=      "polynomial"
 lowerCAmelCase__  					=      "constant"
 lowerCAmelCase__  					=      "constant_with_warmup"
 lowerCAmelCase__  					=      "piecewise_constant"







def    a__       (  a__      ,    a__ = -1						):




     """simple docstring"""


     return LambdaLR(a__      ,    lambda a__						: 1      ,    last_epoch=a__						)







def    a__       (  a__      ,    a__      ,    a__ = -1						):




     """simple docstring"""


     def lr_lambda(a__						):
          if current_step < num_warmup_steps:
               return float(a__						) / float(max(1.0      ,    a__						)						)
          return 1.0

     return LambdaLR(a__      ,    a__      ,    last_epoch=a__						)







def    a__       (  a__      ,    a__      ,    a__ = -1						):




     """simple docstring"""


     __SCREAMING_SNAKE_CASE     =      {}
     __SCREAMING_SNAKE_CASE     =      step_rules.split(""","""						)
     for rule_str in rule_list[:-1]:
          __SCREAMING_SNAKE_CASE   ,			__SCREAMING_SNAKE_CASE     =      rule_str.split(""":"""						)
          __SCREAMING_SNAKE_CASE     =      int(a__						)
          __SCREAMING_SNAKE_CASE     =      float(a__						)
          __SCREAMING_SNAKE_CASE     =      value
     __SCREAMING_SNAKE_CASE     =      float(rule_list[-1]						)

     def create_rules_function(a__      ,    a__						):
          def rule_func(a__						) -> float:
               __SCREAMING_SNAKE_CASE     =      sorted(rules_dict.keys()						)
               for i, sorted_step in enumerate(a__						):
                    if steps < sorted_step:
                         return rules_dict[sorted_steps[i]]
               return last_lr_multiple

          return rule_func

     __SCREAMING_SNAKE_CASE     =      create_rules_function(a__      ,    a__						)

     return LambdaLR(a__      ,    a__      ,    last_epoch=a__						)







def    a__       (  a__      ,    a__      ,    a__      ,    a__=-1						):




     """simple docstring"""


     def lr_lambda(a__						):
          if current_step < num_warmup_steps:
               return float(a__						) / float(max(1      ,    a__						)						)
          return max(
              0.0      ,    float(num_training_steps - current_step						) / float(max(1      ,    num_training_steps - num_warmup_steps						)						)						)

     return LambdaLR(a__      ,    a__      ,    a__						)







def    a__       (  a__      ,    a__      ,    a__      ,    a__ = 0.5      ,    a__ = -1						):




     """simple docstring"""


     def lr_lambda(a__						):
          if current_step < num_warmup_steps:
               return float(a__						) / float(max(1      ,    a__						)						)
          __SCREAMING_SNAKE_CASE     =      float(current_step - num_warmup_steps						) / float(max(1      ,    num_training_steps - num_warmup_steps						)						)
          return max(0.0      ,    0.5 * (1.0 + math.cos(math.pi * float(a__						) * 2.0 * progress						))						)

     return LambdaLR(a__      ,    a__      ,    a__						)







def    a__       (  a__      ,    a__      ,    a__      ,    a__ = 1      ,    a__ = -1						):




     """simple docstring"""


     def lr_lambda(a__						):
          if current_step < num_warmup_steps:
               return float(a__						) / float(max(1      ,    a__						)						)
          __SCREAMING_SNAKE_CASE     =      float(current_step - num_warmup_steps						) / float(max(1      ,    num_training_steps - num_warmup_steps						)						)
          if progress >= 1.0:
               return 0.0
          return max(0.0      ,    0.5 * (1.0 + math.cos(math.pi * ((float(a__						) * progress) % 1.0)						))						)

     return LambdaLR(a__      ,    a__      ,    a__						)







def    a__       (  a__      ,    a__      ,    a__      ,    a__=1E-7      ,    a__=1.0      ,    a__=-1						):




     """simple docstring"""


     __SCREAMING_SNAKE_CASE     =      optimizer.defaults["""lr"""]
     if not (lr_init > lr_end):
          raise ValueError(F'lr_end ({lr_end}) must be be smaller than initial lr ({lr_init})'						)

     def lr_lambda(a__						):
          if current_step < num_warmup_steps:
               return float(a__						) / float(max(1      ,    a__						)						)
          elif current_step > num_training_steps:
               return lr_end / lr_init  # as LambdaLR multiplies by lr_init
          else:
               __SCREAMING_SNAKE_CASE     =      lr_init - lr_end
               __SCREAMING_SNAKE_CASE     =      num_training_steps - num_warmup_steps
               __SCREAMING_SNAKE_CASE     =      1 - (current_step - num_warmup_steps) / decay_steps
               __SCREAMING_SNAKE_CASE     =      lr_range * pct_remaining**power + lr_end
               return decay / lr_init  # as LambdaLR multiplies by lr_init

     return LambdaLR(a__      ,    a__      ,    a__						)


UpperCAmelCase :       Optional[Any]			   =     {
    SchedulerType.LINEAR: get_linear_schedule_with_warmup,
    SchedulerType.COSINE: get_cosine_schedule_with_warmup,
    SchedulerType.COSINE_WITH_RESTARTS: get_cosine_with_hard_restarts_schedule_with_warmup,
    SchedulerType.POLYNOMIAL: get_polynomial_decay_schedule_with_warmup,
    SchedulerType.CONSTANT: get_constant_schedule,
    SchedulerType.CONSTANT_WITH_WARMUP: get_constant_schedule_with_warmup,
    SchedulerType.PIECEWISE_CONSTANT: get_piecewise_constant_schedule,
}







def    a__       (  a__      ,    a__      ,    a__ = None      ,    a__ = None      ,    a__ = None      ,    a__ = 1      ,    a__ = 1.0      ,    a__ = -1      ,    ):




     """simple docstring"""


     __SCREAMING_SNAKE_CASE     =      SchedulerType(a__						)
     __SCREAMING_SNAKE_CASE     =      TYPE_TO_SCHEDULER_FUNCTION[name]
     if name == SchedulerType.CONSTANT:
          return schedule_func(a__      ,    last_epoch=a__						)

     if name == SchedulerType.PIECEWISE_CONSTANT:
          return schedule_func(a__      ,    step_rules=a__      ,    last_epoch=a__						)

     # All other schedulers require `num_warmup_steps`
     if num_warmup_steps is None:
          raise ValueError(F'{name} requires `num_warmup_steps`, please provide that argument.'						)

     if name == SchedulerType.CONSTANT_WITH_WARMUP:
          return schedule_func(a__      ,    num_warmup_steps=a__      ,    last_epoch=a__						)

     # All other schedulers require `num_training_steps`
     if num_training_steps is None:
          raise ValueError(F'{name} requires `num_training_steps`, please provide that argument.'						)

     if name == SchedulerType.COSINE_WITH_RESTARTS:
          return schedule_func(
              a__      ,    num_warmup_steps=a__      ,    num_training_steps=a__      ,    num_cycles=a__      ,    last_epoch=a__      ,    )

     if name == SchedulerType.POLYNOMIAL:
          return schedule_func(
              a__      ,    num_warmup_steps=a__      ,    num_training_steps=a__      ,    power=a__      ,    last_epoch=a__      ,    )

     return schedule_func(
         a__      ,    num_warmup_steps=a__      ,    num_training_steps=a__      ,    last_epoch=a__						)






