





import copy
import os
from typing import Union

from ...configuration_utils import PretrainedConfig
from ...models.auto.modeling_auto import MODEL_FOR_CAUSAL_LM_MAPPING_NAMES
from ...utils import logging
from ..auto import CONFIG_MAPPING


__A							:			str         =			logging.get_logger(__name__)

__A							:			Any         =			{
    "salesforce/blip2-opt-2.7b": "https://huggingface.co/salesforce/blip2-opt-2.7b/resolve/main/config.json",
}







class      lowerCamelCase(						__snake_case	):
     '''simple docstring'''


     __magic_name__   =			'blip_2_vision_model'




     def __init__(    self				,     snake_case_=1408				,     snake_case_=6144				,     snake_case_=39				,     snake_case_=16				,     snake_case_=224				,     snake_case_=14				,     snake_case_="gelu"				,     snake_case_=0.0_0001				,     snake_case_=0.0				,     snake_case_=1E-10				,     snake_case_=True				,     **snake_case_				,     ):
         super().__init__(**snake_case_							)

         _A     =							hidden_size
         _A     =							intermediate_size
         _A     =							num_hidden_layers
         _A     =							num_attention_heads
         _A     =							patch_size
         _A     =							image_size
         _A     =							initializer_range
         _A     =							attention_dropout
         _A     =							layer_norm_eps
         _A     =							hidden_act
         _A     =							qkv_bias




     @classmethod
     def        lowerCAmelCase__		(    cls				,     snake_case_				,     **snake_case_							):
         cls._set_token_in_kwargs(snake_case_							)

         _A,			_A     =							cls.get_config_dict(snake_case_				,     **snake_case_							)

         # get the vision config dict if we are loading from Blip2Config
         if config_dict.get('model_type'							) == "blip-2":
             _A     =							config_dict['vision_config']

         if "model_type" in config_dict and hasattr(cls				,     'model_type'							) and config_dict["model_type"] != cls.model_type:
             logger.warning(
                 F"You are using a model of type {config_dict['model_type']} to instantiate a model of type "
                 F"{cls.model_type}. This is not supported for all configurations of models and can yield errors."							)

         return cls.from_dict(snake_case_				,     **snake_case_							)







class      lowerCamelCase(						__snake_case	):
     '''simple docstring'''


     __magic_name__   =			'blip_2_qformer'




     def __init__(    self				,     snake_case_=3_0522				,     snake_case_=768				,     snake_case_=12				,     snake_case_=12				,     snake_case_=3072				,     snake_case_="gelu"				,     snake_case_=0.1				,     snake_case_=0.1				,     snake_case_=512				,     snake_case_=0.02				,     snake_case_=1E-12				,     snake_case_=0				,     snake_case_="absolute"				,     snake_case_=2				,     snake_case_=1408				,     **snake_case_				,     ):
         super().__init__(pad_token_id=snake_case_				,     **snake_case_							)

         _A     =							vocab_size
         _A     =							hidden_size
         _A     =							num_hidden_layers
         _A     =							num_attention_heads
         _A     =							hidden_act
         _A     =							intermediate_size
         _A     =							hidden_dropout_prob
         _A     =							attention_probs_dropout_prob
         _A     =							max_position_embeddings
         _A     =							initializer_range
         _A     =							layer_norm_eps
         _A     =							position_embedding_type
         _A     =							cross_attention_frequency
         _A     =							encoder_hidden_size




     @classmethod
     def        lowerCAmelCase__		(    cls				,     snake_case_				,     **snake_case_							):
         cls._set_token_in_kwargs(snake_case_							)

         _A,			_A     =							cls.get_config_dict(snake_case_				,     **snake_case_							)

         # get the qformer config dict if we are loading from Blip2Config
         if config_dict.get('model_type'							) == "blip-2":
             _A     =							config_dict['qformer_config']

         if "model_type" in config_dict and hasattr(cls				,     'model_type'							) and config_dict["model_type"] != cls.model_type:
             logger.warning(
                 F"You are using a model of type {config_dict['model_type']} to instantiate a model of type "
                 F"{cls.model_type}. This is not supported for all configurations of models and can yield errors."							)

         return cls.from_dict(snake_case_				,     **snake_case_							)







class      lowerCamelCase(						__snake_case	):
     '''simple docstring'''


     __magic_name__   =			'blip-2'
     __magic_name__   =			True




     def __init__(    self				,     snake_case_=None				,     snake_case_=None				,     snake_case_=None				,     snake_case_=32				,     **snake_case_							):
         super().__init__(**snake_case_							)

         if vision_config is None:
             _A     =							{}
             logger.info('vision_config is None. initializing the Blip2VisionConfig with default values.'							)

         if qformer_config is None:
             _A     =							{}
             logger.info('qformer_config is None. Initializing the Blip2QFormerConfig with default values.'							)

         if text_config is None:
             _A     =							{}
             logger.info('text_config is None. Initializing the text config with default values (`OPTConfig`).'							)

         _A     =							BlipaVisionConfig(**snake_case_							)
         _A     =							BlipaQFormerConfig(**snake_case_							)
         _A     =							text_config['model_type'] if 'model_type' in text_config else 'opt'
         _A     =							CONFIG_MAPPING[text_model_type](**snake_case_							)

         _A     =							self.text_config.tie_word_embeddings
         _A     =							self.text_config.is_encoder_decoder

         _A     =							num_query_tokens
         _A     =							self.vision_config.hidden_size
         _A     =							self.text_config.model_type in MODEL_FOR_CAUSAL_LM_MAPPING_NAMES
         _A     =							1.0
         _A     =							0.02




     @classmethod
     def        lowerCAmelCase__		(    cls				,     snake_case_				,     snake_case_				,     snake_case_				,     **snake_case_				,     ):
         return cls(
             vision_config=vision_config.to_dict()				,     qformer_config=qformer_config.to_dict()				,     text_config=text_config.to_dict()				,     **snake_case_				,     )




     def        lowerCAmelCase__		(    self							):
         _A     =							copy.deepcopy(self.__dict__							)
         _A     =							self.vision_config.to_dict()
         _A     =							self.qformer_config.to_dict()
         _A     =							self.text_config.to_dict()
         _A     =							self.__class__.model_type
         return output












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_rag import save_json





def 				__lowerCAmelCase(       _SCREAMING_SNAKE_CASE				)							-> Optional[int]:




    """simple docstring"""




    _A     =							filter(lambda _SCREAMING_SNAKE_CASE				: p.requires_grad							,     model.parameters()				)
    _A     =							sum([np.prod(p.size()				) for p in model_parameters]				)
    return params


__A							:			Union[str, Any]         =			logging.getLogger(__name__)





def 				__lowerCAmelCase(       _SCREAMING_SNAKE_CASE							,     _SCREAMING_SNAKE_CASE				)							-> int:




    """simple docstring"""




    if metric == "rouge2":
        _A     =							'{val_avg_rouge2:.4f}-{step_count}'
    elif metric == "bleu":
        _A     =							'{val_avg_bleu:.4f}-{step_count}'
    elif metric == "em":
        _A     =							'{val_avg_em:.4f}-{step_count}'
    elif metric == "loss":
        _A     =							'{val_avg_loss:.4f}-{step_count}'
    else:
        raise NotImplementedError(
            F"seq2seq callbacks only support rouge2 and bleu, got {metric}, You can make your own by adding to this"
            ' function.'				)

    _A     =							ModelCheckpoint(
        dirpath=_SCREAMING_SNAKE_CASE							,     filename=_SCREAMING_SNAKE_CASE							,     monitor=F"val_{metric}"							,     mode='max'							,     save_top_k=1							,     every_n_epochs=1							,     )

    return checkpoint_callback





def 				__lowerCAmelCase(       _SCREAMING_SNAKE_CASE							,     _SCREAMING_SNAKE_CASE				)							-> Tuple:




    """simple docstring"""




    return EarlyStopping(
        monitor=F"val_{metric}"							,     mode='min' if 'loss' in metric else 'max'							,     patience=_SCREAMING_SNAKE_CASE							,     verbose=_SCREAMING_SNAKE_CASE							,     )







class      lowerCamelCase(						pl.Callback	):
     '''simple docstring'''


     def        lowerCAmelCase__		(    self				,     snake_case_				,     snake_case_							):
         _A     =							{F"lr_group_{i}": param['lr'] for i, param in enumerate(pl_module.trainer.optimizers[0].param_groups							)}
         pl_module.logger.log_metrics(snake_case_							)




     @rank_zero_only
     def        lowerCAmelCase__		(    self				,     snake_case_				,     snake_case_				,     snake_case_				,     snake_case_=True							):
         logger.info(F"***** {type_path} results at step {trainer.global_step:05d} *****"							)
         _A     =							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
         _A     =							Path(pl_module.hparams.output_dir							)
         if type_path == "test":
             _A     =							od / 'test_results.txt'
             _A     =							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.
             _A     =							od / F"{type_path}_results/{trainer.global_step:05d}.txt"
             _A     =							od / F"{type_path}_generations/{trainer.global_step:05d}.txt"
             results_file.parent.mkdir(exist_ok=snake_case_							)
             generations_file.parent.mkdir(exist_ok=snake_case_							)
         with open(snake_case_				,     'a+'							) as writer:
             for key in sorted(snake_case_							):
                 if key in ["log", "progress_bar", "preds"]:
                     continue
                 _A     =							metrics[key]
                 if isinstance(snake_case_				,     torch.Tensor							):
                     _A     =							val.item()
                 _A     =							F"{key}: {val:.6f}\n"
                 writer.write(snake_case_							)

         if not save_generations:
             return

         if "preds" in metrics:
             _A     =							'\n'.join(metrics['preds']							)
             generations_file.open('w+'							).write(snake_case_							)




     @rank_zero_only
     def        lowerCAmelCase__		(    self				,     snake_case_				,     snake_case_							):
         try:
             _A     =							pl_module.model.model.num_parameters()
         except AttributeError:
             _A     =							pl_module.model.num_parameters()

         _A     =							count_trainable_parameters(snake_case_							)
         # 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				,     snake_case_				,     snake_case_							):
         save_json(pl_module.metrics				,     pl_module.metrics_save_path							)
         return self._write_logs(snake_case_				,     snake_case_				,     'test'							)




     @rank_zero_only
     def        lowerCAmelCase__		(    self				,     snake_case_				,     snake_case_							):
         save_json(pl_module.metrics				,     pl_module.metrics_save_path							)
         # Uncommenting this will save val generations
         # return self._write_logs(trainer, pl_module, "valid")





