





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  __UpperCAmelCase(   lowercase_   ):
       _lowerCamelCase				:  Union[str, Any]						=		filter(lambda lowercase_   : p.requires_grad			,     model.parameters()   )
       _lowerCamelCase				:  int						=		sum([np.prod(p.size()   ) for p in model_parameters]   )
       return params


_lowerCamelCase      							=   logging.getLogger(__name__)




def  __UpperCAmelCase(   lowercase_			,     lowercase_   ):
       if metric == "rouge2":
              _lowerCamelCase				:  List[str]						=		'''{val_avg_rouge2:.4f}-{step_count}'''
       elif metric == "bleu":
              _lowerCamelCase				:  str						=		'''{val_avg_bleu:.4f}-{step_count}'''
       elif metric == "em":
              _lowerCamelCase				:  int						=		'''{val_avg_em:.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.'''   )

       _lowerCamelCase				:  Any						=		ModelCheckpoint(
           dirpath=lowercase_			,     filename=lowercase_			,     monitor=F"""val_{metric}"""			,     mode='''max'''			,     save_top_k=3			,     every_n_epochs=1			,     )
       return checkpoint_callback




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







class       __A							(   pl.Callback	):




      """simple docstring"""







      def 					__snake_case      (		self   ,				a__   ,				a__):



             """simple docstring"""





             _lowerCamelCase				:  Optional[int]						=		{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 					__snake_case      (		self   ,				a__   ,				a__   ,				a__   ,				a__=True):



             """simple docstring"""





             logger.info(F"""***** {type_path} results at step {trainer.global_step:05d} *****""")
             _lowerCamelCase				:  List[str]						=		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
             _lowerCamelCase				:  str						=		Path(pl_module.hparams.output_dir)
             if type_path == "test":
                    _lowerCamelCase				:  Dict						=		od / '''test_results.txt'''
                    _lowerCamelCase				:  Union[str, Any]						=		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.
                    _lowerCamelCase				:  Dict						=		od / F"""{type_path}_results/{trainer.global_step:05d}.txt"""
                    _lowerCamelCase				:  Optional[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
                           _lowerCamelCase				:  Optional[int]						=		metrics[key]
                           if isinstance(a__   ,				torch.Tensor):
                                  _lowerCamelCase				:  int						=		val.item()
                           _lowerCamelCase				:  Any						=		F"""{key}: {val:.6f}\n"""
                           writer.write(a__)

             if not save_generations:
                    return

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


      @rank_zero_only
      def 					__snake_case      (		self   ,				a__   ,				a__):



             """simple docstring"""





             try:
                    _lowerCamelCase				:  Union[str, Any]						=		pl_module.model.model.num_parameters()
             except AttributeError:
                    _lowerCamelCase				:  Union[str, Any]						=		pl_module.model.num_parameters()

             _lowerCamelCase				:  List[str]						=		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 					__snake_case      (		self   ,				a__   ,				a__):



             """simple docstring"""





             save_json(pl_module.metrics   ,				pl_module.metrics_save_path)
             return self._write_logs(a__   ,				a__   ,				'''test''')



      @rank_zero_only
      def 					__snake_case      (		self   ,				a__   ,				a__):



             """simple docstring"""





             save_json(pl_module.metrics   ,				pl_module.metrics_save_path)
             # Uncommenting this will save val generations
             # return self._write_logs(trainer, pl_module, "valid")








import shutil
import tempfile
import unittest

import numpy as np
import pytest

from transformers.testing_utils import require_vision
from transformers.utils import is_vision_available


if is_vision_available():
       from PIL import Image

       from transformers import (
           AutoProcessor,
           BertTokenizerFast,
           BlipImageProcessor,
           GPTaTokenizer,
           InstructBlipProcessor,
           PreTrainedTokenizerFast,
       )







@require_vision
class       __A							(   unittest.TestCase	):




      """simple docstring"""







      def 					__snake_case      (		self):



             """simple docstring"""





             _lowerCamelCase				:  List[Any]						=		tempfile.mkdtemp()

             _lowerCamelCase				:  List[str]						=		BlipImageProcessor()
             _lowerCamelCase				:  Union[str, Any]						=		GPTaTokenizer.from_pretrained('''hf-internal-testing/tiny-random-GPT2Model''')
             _lowerCamelCase				:  Any						=		BertTokenizerFast.from_pretrained('''hf-internal-testing/tiny-random-bert''')

             _lowerCamelCase				:  Any						=		InstructBlipProcessor(a__   ,				a__   ,				a__)

             processor.save_pretrained(self.tmpdirname)


      def 					__snake_case      (		self   ,				**a__):



             """simple docstring"""





             return AutoProcessor.from_pretrained(self.tmpdirname   ,				**a__).tokenizer


      def 					__snake_case      (		self   ,				**a__):



             """simple docstring"""





             return AutoProcessor.from_pretrained(self.tmpdirname   ,				**a__).image_processor


      def 					__snake_case      (		self   ,				**a__):



             """simple docstring"""





             return AutoProcessor.from_pretrained(self.tmpdirname   ,				**a__).qformer_tokenizer


      def 					__snake_case      (		self):



             """simple docstring"""





             shutil.rmtree(self.tmpdirname)


      def 					__snake_case      (		self):



             """simple docstring"""





             _lowerCamelCase				:  List[Any]						=		[np.random.randint(255   ,				size=(3, 30, 400)   ,				dtype=np.uinta)]

             _lowerCamelCase				:  List[str]						=		[Image.fromarray(np.moveaxis(a__   ,				0   ,				-1)) for x in image_inputs]

             return image_inputs


      def 					__snake_case      (		self):



             """simple docstring"""





             _lowerCamelCase				:  Optional[Any]						=		InstructBlipProcessor(
                 tokenizer=self.get_tokenizer()   ,				image_processor=self.get_image_processor()   ,				qformer_tokenizer=self.get_qformer_tokenizer()   ,				)
             processor.save_pretrained(self.tmpdirname)

             _lowerCamelCase				:  Optional[int]						=		self.get_tokenizer(bos_token='''(BOS)'''   ,				eos_token='''(EOS)''')
             _lowerCamelCase				:  str						=		self.get_image_processor(do_normalize=a__   ,				padding_value=1.0)

             _lowerCamelCase				:  Tuple						=		InstructBlipProcessor.from_pretrained(
                 self.tmpdirname   ,				bos_token='''(BOS)'''   ,				eos_token='''(EOS)'''   ,				do_normalize=a__   ,				padding_value=1.0)

             self.assertEqual(processor.tokenizer.get_vocab()   ,				tokenizer_add_kwargs.get_vocab())
             self.assertIsInstance(processor.tokenizer   ,				a__)

             self.assertEqual(processor.image_processor.to_json_string()   ,				image_processor_add_kwargs.to_json_string())
             self.assertIsInstance(processor.image_processor   ,				a__)
             self.assertIsInstance(processor.qformer_tokenizer   ,				a__)


      def 					__snake_case      (		self):



             """simple docstring"""





             _lowerCamelCase				:  List[str]						=		self.get_image_processor()
             _lowerCamelCase				:  Optional[int]						=		self.get_tokenizer()
             _lowerCamelCase				:  Any						=		self.get_qformer_tokenizer()

             _lowerCamelCase				:  Dict						=		InstructBlipProcessor(
                 tokenizer=a__   ,				image_processor=a__   ,				qformer_tokenizer=a__)

             _lowerCamelCase				:  List[Any]						=		self.prepare_image_inputs()

             _lowerCamelCase				:  List[str]						=		image_processor(a__   ,				return_tensors='''np''')
             _lowerCamelCase				:  List[Any]						=		processor(images=a__   ,				return_tensors='''np''')

             for key in input_feat_extract.keys():
                    self.assertAlmostEqual(input_feat_extract[key].sum()   ,				input_processor[key].sum()   ,				delta=1e-2)


      def 					__snake_case      (		self):



             """simple docstring"""





             _lowerCamelCase				:  Any						=		self.get_image_processor()
             _lowerCamelCase				:  Dict						=		self.get_tokenizer()
             _lowerCamelCase				:  int						=		self.get_qformer_tokenizer()

             _lowerCamelCase				:  Tuple						=		InstructBlipProcessor(
                 tokenizer=a__   ,				image_processor=a__   ,				qformer_tokenizer=a__)

             _lowerCamelCase				:  List[Any]						=		'''lower newer'''

             _lowerCamelCase				:  Any						=		processor(text=a__)

             _lowerCamelCase				:  Optional[int]						=		tokenizer(a__   ,				return_token_type_ids=a__)
             _lowerCamelCase				:  Optional[Any]						=		qformer_tokenizer(a__   ,				return_token_type_ids=a__)

             for key in encoded_tokens.keys():
                    self.assertListEqual(encoded_tokens[key]   ,				encoded_processor[key])

             for key in encoded_tokens_qformer.keys():
                    self.assertListEqual(encoded_tokens_qformer[key]   ,				encoded_processor['''qformer_''' + key])


      def 					__snake_case      (		self):



             """simple docstring"""





             _lowerCamelCase				:  List[Any]						=		self.get_image_processor()
             _lowerCamelCase				:  Optional[int]						=		self.get_tokenizer()
             _lowerCamelCase				:  Optional[Any]						=		self.get_qformer_tokenizer()

             _lowerCamelCase				:  List[Any]						=		InstructBlipProcessor(
                 tokenizer=a__   ,				image_processor=a__   ,				qformer_tokenizer=a__)

             _lowerCamelCase				:  List[Any]						=		'''lower newer'''
             _lowerCamelCase				:  Tuple						=		self.prepare_image_inputs()

             _lowerCamelCase				:  Tuple						=		processor(text=a__   ,				images=a__)

             self.assertListEqual(
                 list(inputs.keys())   ,				['''input_ids''', '''attention_mask''', '''qformer_input_ids''', '''qformer_attention_mask''', '''pixel_values''']   ,				)

             # test if it raises when no input is passed
             with pytest.raises(a__):
                    processor()


      def 					__snake_case      (		self):



             """simple docstring"""





             _lowerCamelCase				:  int						=		self.get_image_processor()
             _lowerCamelCase				:  List[Any]						=		self.get_tokenizer()
             _lowerCamelCase				:  List[Any]						=		self.get_qformer_tokenizer()

             _lowerCamelCase				:  Optional[Any]						=		InstructBlipProcessor(
                 tokenizer=a__   ,				image_processor=a__   ,				qformer_tokenizer=a__)

             _lowerCamelCase				:  List[Any]						=		[[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

             _lowerCamelCase				:  Tuple						=		processor.batch_decode(a__)
             _lowerCamelCase				:  str						=		tokenizer.batch_decode(a__)

             self.assertListEqual(a__   ,				a__)



      def 					__snake_case      (		self):



             """simple docstring"""





             _lowerCamelCase				:  List[str]						=		self.get_image_processor()
             _lowerCamelCase				:  Dict						=		self.get_tokenizer()
             _lowerCamelCase				:  str						=		self.get_qformer_tokenizer()

             _lowerCamelCase				:  str						=		InstructBlipProcessor(
                 tokenizer=a__   ,				image_processor=a__   ,				qformer_tokenizer=a__)

             _lowerCamelCase				:  str						=		'''lower newer'''
             _lowerCamelCase				:  str						=		self.prepare_image_inputs()

             _lowerCamelCase				:  List[Any]						=		processor(text=a__   ,				images=a__)

             self.assertListEqual(
                 list(inputs.keys())   ,				['''input_ids''', '''attention_mask''', '''qformer_input_ids''', '''qformer_attention_mask''', '''pixel_values''']   ,				)

