






"""simple docstring"""





from typing import List, Optional, Tuple

from tokenizers import pre_tokenizers

from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import logging
from .tokenization_clip import CLIPTokenizer


UpperCamelCase_						:					Optional[Any]      				=			logging.get_logger(__name__)

UpperCamelCase_						:					List[str]      				=			{"""vocab_file""": """vocab.json""", """merges_file""": """merges.txt""", """tokenizer_file""": """tokenizer.json"""}

UpperCamelCase_						:					List[str]      				=			{
    """vocab_file""": {
        """openai/clip-vit-base-patch32""": """https://huggingface.co/openai/clip-vit-base-patch32/resolve/main/vocab.json""",
    },
    """merges_file""": {
        """openai/clip-vit-base-patch32""": """https://huggingface.co/openai/clip-vit-base-patch32/resolve/main/merges.txt""",
    },
    """tokenizer_file""": {
        """openai/clip-vit-base-patch32""": (
            """https://huggingface.co/openai/clip-vit-base-patch32/resolve/main/tokenizer.json"""
        ),
    },
}

UpperCamelCase_						:					List[Any]      				=			{
    """openai/clip-vit-base-patch32""": 77,
}


class   _lowercase			(  _snake_case    ):
      _a :  List[str]										= VOCAB_FILES_NAMES
      _a :  int										= PRETRAINED_VOCAB_FILES_MAP
      _a :  Union[str, Any]										= PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      _a :  Optional[Any]										= ['''input_ids''', '''attention_mask''']
      _a :  Optional[int]										= CLIPTokenizer


      def __init__(     self					:						str				, a					:						List[Any]=None				, a					:						List[Any]=None				, a					:						Optional[Any]=None				, a					:						Union[str, Any]="<|endoftext|>"				, a					:						Any="<|startoftext|>"				, a					:						List[Any]="<|endoftext|>"				, a					:						Optional[int]="<|endoftext|>"				, **a					:						Any				, ):





            """simple docstring"""







            super().__init__(
                _a				, _a				, tokenizer_file=_a				, unk_token=_a				, bos_token=_a				, eos_token=_a				, pad_token=_a				, **_a				, )

            if not isinstance(self.backend_tokenizer.pre_tokenizer				, pre_tokenizers.Sequence    ):
                  raise ValueError(
                      '''The `backend_tokenizer` provided does not match the expected format. The CLIP tokenizer has been'''
                      ''' heavily modified from transformers version 4.17.0. You need to convert the tokenizer you are using'''
                      ''' to be compatible with this version.The easiest way to do so is'''
                      ''' `CLIPTokenizerFast.from_pretrained("path_to_local_folder_or_hub_repo, from_slow=True)`. If you want'''
                      ''' to use your existing tokenizer, you will have to revert to a version prior to 4.17.0 of'''
                      ''' transformers.'''    )

            self._wrap_decode_method_backend_tokenizer()


      def _UpperCamelCase    (     self					:						List[str]    ):





            """simple docstring"""







            __snake_case   :     str	      =self.backend_tokenizer.decode

            def new_decode_method(*a					:						List[Any]				, **a					:						Dict    ):
                  __snake_case   :     Optional[int]	      =orig_decode_method(*_a				, **_a    )
                  __snake_case   :     int	      =text.replace(self.backend_tokenizer.model.end_of_word_suffix				, ''' '''    ).strip()
                  return text

            __snake_case   :     Dict	      =new_decode_method


      def _UpperCamelCase    (     self					:						List[str]				, a					:						List[Any]				, a					:						int = None    ):





            """simple docstring"""







            __snake_case   :     List[str]	      =[self.bos_token_id]
            __snake_case   :     Tuple	      =[self.eos_token_id]

            if token_ids_a is None:
                  return bos_token + token_ids_a + eos_token
            return bos_token + token_ids_a + eos_token + eos_token + token_ids_a + eos_token


      def _UpperCamelCase    (     self					:						int				, a					:						Union[str, Any]				, a					:						Dict = None    ):





            """simple docstring"""







            __snake_case   :     int	      =[self.bos_token_id]
            __snake_case   :     int	      =[self.eos_token_id]

            if token_ids_a is None:
                  return len(bos_token + token_ids_a + eos_token    ) * [0]
            return len(bos_token + token_ids_a + eos_token + eos_token + token_ids_a + eos_token    ) * [0]







      def _UpperCamelCase    (     self					:						str				, a					:						Optional[int]				, a					:						int = None    ):





            """simple docstring"""







            __snake_case   :     str	      =self._tokenizer.model.save(_a				, name=_a    )
            return tuple(_a    )



'''simple docstring'''




import logging
from pathlib import Path

import numpy as np
import pytorch_lightning as pl
import torch
from pytorch_lightning.callbacks import EarlyStopping, ModelCheckpoint
from pytorch_lightning.utilities import rank_zero_only

from utils import save_json

def _UpperCamelCase (							SCREAMING_SNAKE_CASE_				):
     lowercase_       :							Optional[Any]  					=       filter(lambda SCREAMING_SNAKE_CASE_				: p.requires_grad	,							model.parameters()				)
     lowercase_       :							Tuple  					=       sum([np.prod(p.size()				) for p in model_parameters]				)
     return params


_A										= logging.getLogger(__name__)


class    UpperCAmelCase__	(							pl.Callback      ):






   """simple docstring"""
   def 		_lowerCamelCase			(self							,					_a							,					_a					)     ->  str:
        lowercase_       :							List[str]  					=       {f'''lr_group_{i}''': param['lr'] for i, param in enumerate(pl_module.trainer.optimizers[0].param_groups					)}
        pl_module.logger.log_metrics(_a					)
   @rank_zero_only
   def 		_lowerCamelCase			(self							,					_a							,					_a							,					_a							,					_a=True					)     ->  None:
        logger.info(f'''***** {type_path} results at step {trainer.global_step:05d} *****'''					)
        lowercase_       :							Union[str, Any]  					=       trainer.callback_metrics
        trainer.logger.log_metrics({k: v for k, v in metrics.items() if k not in ['log', 'progress_bar', 'preds']}					)
        # Log results
        lowercase_       :							Optional[int]  					=       Path(pl_module.hparams.output_dir					)
        if type_path == "test":
             lowercase_       :							Union[str, Any]  					=       od / 'test_results.txt'
             lowercase_       :							Tuple  					=       od / 'test_generations.txt'
        else:
             # this never gets hit. I prefer not to save intermediate generations, and results are in metrics.json
             # If people want this it will be easy enough to add back.
             lowercase_       :							List[str]  					=       od / f'''{type_path}_results/{trainer.global_step:05d}.txt'''
             lowercase_       :							int  					=       od / f'''{type_path}_generations/{trainer.global_step:05d}.txt'''
             results_file.parent.mkdir(exist_ok=_a					)
             generations_file.parent.mkdir(exist_ok=_a					)
        with open(_a							,					'a+'					) as writer:
             for key in sorted(_a					):
                  if key in ["log", "progress_bar", "preds"]:
                       continue
                  lowercase_       :							Any  					=       metrics[key]
                  if isinstance(_a							,					torch.Tensor					):
                       lowercase_       :							Any  					=       val.item()
                  lowercase_       :							List[str]  					=       f'''{key}: {val:.6f}\n'''
                  writer.write(_a					)

        if not save_generations:
             return

        if "preds" in metrics:
             lowercase_       :							Optional[Any]  					=       '\n'.join(metrics['preds']					)
             generations_file.open('w+'					).write(_a					)
   @rank_zero_only
   def 		_lowerCamelCase			(self							,					_a							,					_a					)     ->  Tuple:
        try:
             lowercase_       :							List[Any]  					=       pl_module.model.model.num_parameters()
        except AttributeError:
             lowercase_       :							Dict  					=       pl_module.model.num_parameters()

        lowercase_       :							Optional[Any]  					=       count_trainable_parameters(_a					)
        # mp stands for million parameters
        trainer.logger.log_metrics({'n_params': npars, 'mp': npars / 1e6, 'grad_mp': n_trainable_pars / 1e6}					)
   @rank_zero_only
   def 		_lowerCamelCase			(self							,					_a							,					_a					)     ->  int:
        save_json(pl_module.metrics							,					pl_module.metrics_save_path					)
        return self._write_logs(_a							,					_a							,					'test'					)





   @rank_zero_only
   def 		_lowerCamelCase			(self							,					_a							,					_a					)     ->  List[str]:
        save_json(pl_module.metrics							,					pl_module.metrics_save_path					)
        # Uncommenting this will save val generations
        # return self._write_logs(trainer, pl_module, "valid")



def _UpperCamelCase (							SCREAMING_SNAKE_CASE_	,							SCREAMING_SNAKE_CASE_	,							SCREAMING_SNAKE_CASE_=1	,							SCREAMING_SNAKE_CASE_=False				):
     if metric == "rouge2":
          lowercase_       :							Optional[int]  					=       '{val_avg_rouge2:.4f}-{step_count}'
     elif metric == "bleu":
          lowercase_       :							int  					=       '{val_avg_bleu:.4f}-{step_count}'
     elif metric == "loss":
          lowercase_       :							List[Any]  					=       '{val_avg_loss:.4f}-{step_count}'
     else:
          raise NotImplementedError(
              f'''seq2seq callbacks only support rouge2, bleu and loss, got {metric}, You can make your own by adding to'''
              ' this function.'				)

     lowercase_       :							Tuple  					=       ModelCheckpoint(
         dirpath=SCREAMING_SNAKE_CASE_	,							filename=SCREAMING_SNAKE_CASE_	,							monitor=f'''val_{metric}'''	,							mode='min' if 'loss' in metric else 'max'	,							save_top_k=SCREAMING_SNAKE_CASE_	,							)
     return checkpoint_callback






def _UpperCamelCase (							SCREAMING_SNAKE_CASE_	,							SCREAMING_SNAKE_CASE_				):
     return EarlyStopping(
         monitor=f'''val_{metric}'''	,							mode='min' if 'loss' in metric else 'max'	,							patience=SCREAMING_SNAKE_CASE_	,							verbose=SCREAMING_SNAKE_CASE_	,							)




