



"""simple docstring"""




import argparse
import gc
import os
import sys
from pathlib import Path
from typing import List  # noqa: F401

import pytorch_lightning as pl
import torch
from finetune import SummarizationModule, TranslationModule
from finetune import main as ft_main
from make_student import create_student_by_copying_alternating_layers, get_layers_to_supervise
from torch import nn

from transformers import AutoModelForSeqaSeqLM, MBartTokenizer, TaForConditionalGeneration
from transformers.models.bart.modeling_bart import shift_tokens_right
from utils import calculate_bleu, check_output_dir, freeze_params, label_smoothed_nll_loss, use_task_specific_params


# need the parent dir module
sys.path.insert(2, str(Path(__file__).resolve().parents[1]))
from lightning_base import generic_train  # noqa


class 	a__		(    SCREAMING_SNAKE_CASE__  ):
   _lowerCamelCase       				=					['loss', 'ce_loss', 'mlm_loss', 'hid_loss_enc', 'hid_loss_dec']


   def __init__(				self   :				Dict,  lowerCAmelCase   :				Any      )     ->							Optional[int]:
         assert Path(hparams.data_dir      ).exists()
         lowercase	:   str											= Path(hparams.output_dir      )
         self.output_dir.mkdir(exist_ok=lowerCAmelCase      )

         lowercase	:   List[Any]											= self.output_dir.joinpath('student'      )

         lowercase	:   Optional[Any]											= str(lowerCAmelCase      )  # Tell lightning we are training the student
         lowercase	:   List[Any]											= AutoModelForSeqaSeqLM.from_pretrained(hparams.teacher      ).eval()
         use_task_specific_params(lowerCAmelCase,  hparams.task      )  # We copy good generation parameters to student by default
         if hparams.student is not None:
               lowercase	:   List[Any]											= AutoModelForSeqaSeqLM.from_pretrained(hparams.student      )
               use_task_specific_params(lowerCAmelCase,  hparams.task      )
               lowercase						,     lowercase	:   Tuple											= None, None
         else:
               lowercase						,     lowercase						,     lowercase	:   Optional[Any]											= create_student_by_copying_alternating_layers(
                   lowerCAmelCase,  e=hparams.student_encoder_layers,  d=hparams.student_decoder_layers,  save_path=lowerCAmelCase      )

         if hparams.length_penalty != -1:
               lowercase	:   Tuple											= hparams.length_penalty
         lowercase	:   List[str]											= hparams.teacher  # Use teacher's tokenizer
         super().__init__(lowerCAmelCase,  model=lowerCAmelCase,  config=student.config      )
         assert student.config.model_type == teacher.config.model_type, (
             f'''teacher, student model types should be the same, got {student.config.model_type} !='''
             f''' {teacher.config.model_type}'''
         )

         if student.config.model_type == "t5":
               lowercase	:   Union[str, Any]											= len(student.get_encoder().block      )
               lowercase	:   List[Any]											= len(student.get_decoder().block      )
               lowercase	:   int											= len(teacher.get_encoder().block      )
               lowercase	:   List[Any]											= len(teacher.get_decoder().block      )
         else:
               lowercase	:   List[str]											= student.config.encoder_layers
               lowercase	:   Any											= student.config.decoder_layers
               lowercase	:   Dict											= teacher.config.encoder_layers
               lowercase	:   Optional[Any]											= teacher.config.decoder_layers

         lowercase	:   Optional[int]											= not (hparams.student is None or hparams.teacher == hparams.student)
         lowercase	:   Dict											= (not self.different_base_models) and hparams.alpha_hid > 0
         lowercase	:   Optional[int]											= self.different_base_models or (student_encoder_layers != teacher_encoder_layers)
         # self.different_encoder determines whether we need to run the teacher encoder
         lowercase	:   Union[str, Any]											= teacher
         freeze_params(self.teacher      )

         if not self.different_encoder:  # To save RAM, delete teacher encoder and freeze student encoder.
               try:
                     del self.teacher.model.encoder
               except AttributeError:  # T5
                     del self.teacher.encoder

         if e_layer_ids is None:
               lowercase	:   str											= list(range(lowerCAmelCase      )      )
         if d_layer_ids is None:
               lowercase	:   Optional[Any]											= list(range(lowerCAmelCase      )      )

         lowercase						,     lowercase	:   Tuple											= e_layer_ids, d_layer_ids  # type: List[int], List[int]

         if self.do_calc_hidden_loss:  # Intermediate supervision: Decide which layers to supervise
               if hparams.supervise_forward:
                     lowercase	:   Dict											= get_layers_to_supervise(
                         n_student=len(self.e_layer_ids      ),  n_teacher=lowerCAmelCase      )
                     lowercase	:   Any											= get_layers_to_supervise(
                         n_student=len(self.d_layer_ids      ),  n_teacher=lowerCAmelCase      )
               else:  # student layer should emulate hidden states of the teacher layer it was copied from
                     lowercase	:   Union[str, Any]											= self.e_layer_ids
                     lowercase	:   Optional[int]											= self.d_layer_ids
         else:
               lowercase	:   List[str]											= None
               lowercase	:   List[Any]											= None

         lowercase	:   int											= nn.KLDivLoss(reduction='batchmean'      )
         lowercase	:   List[Any]											= 2.0
         lowercase	:   Tuple											= hparams.alpha_mlm
         lowercase	:   List[str]											= hparams.alpha_ce
         lowercase	:   Optional[int]											= hparams.alpha_hid
         gc.collect()
         torch.cuda.empty_cache()


   def 							lowercase					(				self   :				Optional[Any],  lowerCAmelCase   :				List[Any],  lowerCAmelCase   :				Union[str, Any],  lowerCAmelCase   :				Tuple      )     ->							int:
         lowercase	:   str											= mask[:, :, None].expand_as(lowerCAmelCase      )
         lowercase	:   List[Any]											= s_logits.size(-1      )
         lowercase	:   str											= torch.masked_select(lowerCAmelCase,  lowerCAmelCase      )  # (bs * seq_length * voc_size) modulo the 1s in mask
         lowercase	:   List[str]											= torch.masked_select(lowerCAmelCase,  lowerCAmelCase      )  # (bs * seq_length * voc_size) modulo the 1s in mask
         lowercase	:   Dict											= s_logits_slct.view(-1,  lowerCAmelCase      )  # (bs * seq_length, voc_size) modulo the 1s in mask
         lowercase	:   Optional[int]											= t_logits_slct.view(-1,  lowerCAmelCase      )  # (bs * seq_length, voc_size) modulo the 1s in mask
         assert t_logits_slct.size() == s_logits_slct.size()
         lowercase	:   Union[str, Any]											= (
             self.ce_loss_fct(
                 nn.functional.log_softmax(s_logits_slct / self.temperature,  dim=-1      ),  nn.functional.softmax(t_logits_slct / self.temperature,  dim=-1      ),  )
             * (self.temperature) ** 2
         )
         return loss_ce


   @staticmethod
   def 							lowercase					(				lowerCAmelCase   :				List[Any],  lowerCAmelCase   :				Union[str, Any]      )     ->							Optional[Any]:
         SummarizationModule.add_model_specific_args(lowerCAmelCase,  lowerCAmelCase      )
         add_distill_args(lowerCAmelCase      )
         return parser


   def 							lowercase					(				self   :				Dict,  lowerCAmelCase   :				dict      )     ->							tuple:
         lowercase	:   Union[str, Any]											= self.tokenizer.pad_token_id
         lowercase						,     lowercase						,     lowercase	:   List[str]											= batch['input_ids'], batch['attention_mask'], batch['labels']
         if isinstance(self.model,  lowerCAmelCase      ):
               lowercase	:   Union[str, Any]											= self.model._shift_right(lowerCAmelCase      )
         else:
               lowercase	:   List[Any]											= shift_tokens_right(lowerCAmelCase,  lowerCAmelCase      )

         # noinspection PyCallingNonCallable
         lowercase	:   Tuple											= self(
             lowerCAmelCase,  attention_mask=lowerCAmelCase,  decoder_input_ids=lowerCAmelCase,  output_hidden_states=self.do_calc_hidden_loss,  output_attentions=lowerCAmelCase,  use_cache=lowerCAmelCase,  )
         lowercase	:   Dict											= student_outputs['logits']

         # Same cross entropy vs. label smoothing logic as finetune.py
         assert lm_logits.shape[-1] == self.model.config.vocab_size
         if self.hparams.label_smoothing == 0:
               # Same behavior as modeling_bart.py, besides ignoring pad_token_id
               lowercase	:   Union[str, Any]											= nn.CrossEntropyLoss(ignore_index=lowerCAmelCase      )
               lowercase	:   Any											= loss_fct(lm_logits.view(-1,  lm_logits.shape[-1]      ),  labels.view(-1      )      )
         else:
               lowercase	:   Union[str, Any]											= nn.functional.log_softmax(lowerCAmelCase,  dim=-1      )
               lowercase						,     lowercase	:   int											= label_smoothed_nll_loss(
                   lowerCAmelCase,  lowerCAmelCase,  self.hparams.label_smoothing,  ignore_index=lowerCAmelCase      )

         def zero_tensor():
               return torch.tensor(0.0      ).type_as(lowerCAmelCase      )

         lowercase	:   Tuple											= student_outputs[
             'encoder_last_hidden_state'
         ]  # use this unless self.different_base_models
         lowercase						,     lowercase	:   List[Any]											= zero_tensor(), zero_tensor()
         if self.different_encoder:  # compute encoder hidden state loss
               lowercase	:   Optional[Any]											= self.teacher.get_encoder()(
                   lowerCAmelCase,  attention_mask=lowerCAmelCase,  output_hidden_states=self.do_calc_hidden_loss,  )
               if self.different_base_models:
                     lowercase	:   List[str]											= all_teacher_encoder_outputs['last_hidden_state']
               elif self.do_calc_hidden_loss:
                     lowercase	:   List[Any]											= self.calc_hidden_loss(
                         lowerCAmelCase,  student_outputs['encoder_hidden_states'],  all_teacher_encoder_outputs['hidden_states'],  self.e_matches,  normalize_hidden=self.hparams.normalize_hidden,  )

         lowercase	:   List[str]											= self.teacher(
             lowerCAmelCase,  attention_mask=lowerCAmelCase,  encoder_outputs=(teacher_enc_outputs,),  decoder_input_ids=lowerCAmelCase,  output_hidden_states=self.do_calc_hidden_loss,  use_cache=lowerCAmelCase,  )
         lowercase	:   int											= decoder_input_ids.ne(lowerCAmelCase      )
         lowercase	:   int											= self.calc_ce_loss(lowerCAmelCase,  lowerCAmelCase,  teacher_outputs['logits']      )
         if self.do_calc_hidden_loss:  # Intermediate supervision of decoder hidden states
               lowercase	:   int											= self.calc_hidden_loss(
                   lowerCAmelCase,  student_outputs['decoder_hidden_states'],  teacher_outputs['decoder_hidden_states'],  self.d_matches,  normalize_hidden=self.hparams.normalize_hidden,  )

         lowercase	:   int											= (
             self.alpha_ce * loss_ce
             + self.alpha_mlm * student_lm_loss
             + self.hparams.alpha_hid * (hid_loss_enc + hid_loss_dec)
         )
         return blended_loss, loss_ce, student_lm_loss, hid_loss_enc, hid_loss_dec






   @staticmethod
   def 							lowercase					(				lowerCAmelCase   :				Optional[int],  lowerCAmelCase   :				List[str],  lowerCAmelCase   :				int,  lowerCAmelCase   :				Tuple,  lowerCAmelCase   :				List[str]      )     ->							int:
         lowercase	:   List[str]											= 'expected list or tuple for hidden_states, got tensor of shape: '
         assert not isinstance(lowerCAmelCase,  torch.Tensor      ), f'''{msg}{hidden_states.shape}'''
         assert not isinstance(lowerCAmelCase,  torch.Tensor      ), f'''{msg}{hidden_states_T.shape}'''
         lowercase	:   List[str]											= attention_mask.to(hidden_states[0]      )
         lowercase	:   str											= mask.sum() * hidden_states[0].size(-1      )
         lowercase	:   Optional[Any]											= torch.stack([hidden_states[i] for i in range(len(lowerCAmelCase      )      )]      )
         lowercase	:   str											= torch.stack([hidden_states_T[j] for j in matches]      )
         assert student_states.shape == teacher_states.shape, f'''{student_states.shape} != {teacher_states.shape}'''
         if normalize_hidden:
               lowercase	:   Tuple											= nn.functional.layer_norm(lowerCAmelCase,  student_states.shape[1:]      )
               lowercase	:   Dict											= nn.functional.layer_norm(lowerCAmelCase,  teacher_states.shape[1:]      )
         lowercase	:   Dict											= nn.functional.mse_loss(lowerCAmelCase,  lowerCAmelCase,  reduction='none'      )
         lowercase	:   Tuple											= (mse * mask.unsqueeze(0      ).unsqueeze(-1      )).sum() / valid_count
         return masked_mse




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






      '''simple docstring'''


      parser.add_argument('--teacher'						,   type=_UpperCAmelCase					)
      parser.add_argument('--alpha_ce'						,   default=0.8						,   type=_UpperCAmelCase					)
      parser.add_argument('--alpha_mlm'						,   default=0.2						,   type=_UpperCAmelCase					)
      parser.add_argument('--alpha_hid'						,   default=0.0						,   type=_UpperCAmelCase						,   required=_UpperCAmelCase					)
      parser.add_argument('--student'						,   type=_UpperCAmelCase						,   required=_UpperCAmelCase					)
      parser.add_argument('--student_decoder_layers'						,   default=12						,   type=_UpperCAmelCase						,   required=_UpperCAmelCase					)
      parser.add_argument('--student_encoder_layers'						,   default=12						,   type=_UpperCAmelCase						,   required=_UpperCAmelCase					)
      parser.add_argument('--no_teacher'						,   action='store_true'						,   default=_UpperCAmelCase					)
      parser.add_argument('--length_penalty'						,   type=_UpperCAmelCase						,   default=-1					)
      parser.add_argument('--supervise_forward'						,   action='store_true'						,   default=_UpperCAmelCase					)
      parser.add_argument('--normalize_hidden'						,   action='store_true'						,   default=_UpperCAmelCase					)


class 	a__		(    SCREAMING_SNAKE_CASE__  ):
   _lowerCamelCase       				=					'translation'
   _lowerCamelCase       				=					['bleu']
   _lowerCamelCase       				=					'bleu'


   def __init__(				self   :				Dict,  lowerCAmelCase   :				Tuple,  **lowerCAmelCase   :				Optional[Any]      )     ->							List[Any]:
         super().__init__(lowerCAmelCase,  **lowerCAmelCase      )
         assert hparams.src_lang is not None
         assert hparams.tgt_lang is not None
         lowercase	:   Dict											= hparams.src_lang
         lowercase	:   Optional[Any]											= hparams.tgt_lang
         if self.model.config.decoder_start_token_id is None and isinstance(self.tokenizer,  lowerCAmelCase      ):
               lowercase	:   Dict											= self.tokenizer.lang_code_to_id[hparams.tgt_lang]


   def 							lowercase					(				self   :				Any,  lowerCAmelCase   :				List[Any],  lowerCAmelCase   :				int      )     ->							dict:
         return calculate_bleu(lowerCAmelCase,  lowerCAmelCase      )






   @staticmethod
   def 							lowercase					(				lowerCAmelCase   :				Optional[Any],  lowerCAmelCase   :				Tuple      )     ->							List[str]:
         TranslationModule.add_model_specific_args(lowerCAmelCase,  lowerCAmelCase      )
         add_distill_args(lowerCAmelCase      )
         return parser




def 	lowercase__  (      _UpperCAmelCase					)  ->      int:






      '''simple docstring'''


      if args.no_teacher:
            lowercase	:   List[Any]											= TranslationModule if 'translation' in args.task else SummarizationModule
      else:  # DISTILL WITH TEACHER
            lowercase	:   List[Any]											= TranslationDistiller if 'translation' in args.task else SummarizationDistiller
      lowercase	:   str											= module_cls.__name__
      print(f'''using module {args.setup_cls}'''					)
      lowercase	:   int											= module_cls(_UpperCAmelCase					)
      return model


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






      '''simple docstring'''


      Path(args.output_dir					).mkdir(exist_ok=_UpperCAmelCase					)
      check_output_dir(_UpperCAmelCase						,   expected_items=3					)

      lowercase	:   Union[str, Any]											= create_module(_UpperCAmelCase					)
      return ft_main(_UpperCAmelCase						,   model=_UpperCAmelCase					)


if __name__ == "__main__":
     _UpperCamelCase:					Union[str, Any]     				=	argparse.ArgumentParser()
     _UpperCamelCase:					Tuple     				=	pl.Trainer.add_argparse_args(parser)
     _UpperCamelCase:					Union[str, Any]     				=	SummarizationDistiller.add_model_specific_args(parser, os.getcwd())
     _UpperCamelCase:					List[str]     				=	parser.parse_args()

     distill_main(args)





"""simple docstring"""




import unittest
from pathlib import Path

from transformers import is_vision_available, load_tool
from transformers.testing_utils import get_tests_dir

from .test_tools_common import ToolTesterMixin


if is_vision_available():
     from PIL import Image


class 	a__		(    unittest.TestCase,	SCREAMING_SNAKE_CASE__  ):


   def 							lowercase					(				self   :				Dict      )     ->							List[str]:
         lowercase	:   str											= load_tool('image-captioning'      )
         self.tool.setup()
         lowercase	:   List[str]											= load_tool('image-captioning',  remote=lowerCAmelCase      )


   def 							lowercase					(				self   :				str      )     ->							int:
         lowercase	:   Union[str, Any]											= Image.open(Path(get_tests_dir('fixtures/tests_samples/COCO'      )      ) / '000000039769.png'      )
         lowercase	:   List[str]											= self.tool(lowerCAmelCase      )
         self.assertEqual(lowerCAmelCase,  'two cats sleeping on a couch'      )


   def 							lowercase					(				self   :				str      )     ->							List[str]:
         lowercase	:   str											= Image.open(Path(get_tests_dir('fixtures/tests_samples/COCO'      )      ) / '000000039769.png'      )
         lowercase	:   Tuple											= self.remote_tool(lowerCAmelCase      )
         self.assertEqual(lowerCAmelCase,  'two cats sleeping on a couch'      )


   def 							lowercase					(				self   :				Tuple      )     ->							Union[str, Any]:
         lowercase	:   Tuple											= Image.open(Path(get_tests_dir('fixtures/tests_samples/COCO'      )      ) / '000000039769.png'      )
         lowercase	:   Union[str, Any]											= self.tool(image=lowerCAmelCase      )
         self.assertEqual(lowerCAmelCase,  'two cats sleeping on a couch'      )






   def 							lowercase					(				self   :				str      )     ->							Optional[int]:
         lowercase	:   List[str]											= Image.open(Path(get_tests_dir('fixtures/tests_samples/COCO'      )      ) / '000000039769.png'      )
         lowercase	:   str											= self.remote_tool(image=lowerCAmelCase      )
         self.assertEqual(lowerCAmelCase,  'two cats sleeping on a couch'      )
