




'''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 						lowerCAmelCase_   ( lowerCamelCase_   ):







  '''simple docstring'''







  lowerCAmelCase_				:					List[Any]				     =						["""loss""", """ce_loss""", """mlm_loss""", """hid_loss_enc""", """hid_loss_dec"""]


  def __init__(						self    :	List[str]						,		_UpperCAmelCase    :	Optional[int] ):






   """simple docstring"""






   assert Path(hparams.data_dir ).exists()
   UpperCAmelCase__										=    Path(hparams.output_dir )
   self.output_dir.mkdir(exist_ok=_UpperCAmelCase )

   UpperCAmelCase__										=    self.output_dir.joinpath("""student""" )

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

   if hparams.length_penalty != -1:
    UpperCAmelCase__										=    hparams.length_penalty
   UpperCAmelCase__										=    hparams.teacher  # Use teacher's tokenizer
   super().__init__(_UpperCAmelCase						,		model=_UpperCAmelCase						,		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":
    UpperCAmelCase__										=    len(student.get_encoder().block )
    UpperCAmelCase__										=    len(student.get_decoder().block )
    UpperCAmelCase__										=    len(teacher.get_encoder().block )
    UpperCAmelCase__										=    len(teacher.get_decoder().block )
   else:
    UpperCAmelCase__										=    student.config.encoder_layers
    UpperCAmelCase__										=    student.config.decoder_layers
    UpperCAmelCase__										=    teacher.config.encoder_layers
    UpperCAmelCase__										=    teacher.config.decoder_layers

   UpperCAmelCase__										=    not (hparams.student is None or hparams.teacher == hparams.student)
   UpperCAmelCase__										=    (not self.different_base_models) and hparams.alpha_hid > 0
   UpperCAmelCase__										=    self.different_base_models or (student_encoder_layers != teacher_encoder_layers)
   # self.different_encoder determines whether we need to run the teacher encoder
   UpperCAmelCase__										=    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:
    UpperCAmelCase__										=    list(range(_UpperCAmelCase ) )
   if d_layer_ids is None:
    UpperCAmelCase__										=    list(range(_UpperCAmelCase ) )

   UpperCAmelCase__							,			UpperCAmelCase__										=    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:
     UpperCAmelCase__										=    get_layers_to_supervise(
         n_student=len(self.e_layer_ids )						,		n_teacher=_UpperCAmelCase )
     UpperCAmelCase__										=    get_layers_to_supervise(
         n_student=len(self.d_layer_ids )						,		n_teacher=_UpperCAmelCase )
    else:  # student layer should emulate hidden states of the teacher layer it was copied from
     UpperCAmelCase__										=    self.e_layer_ids
     UpperCAmelCase__										=    self.d_layer_ids
   else:
    UpperCAmelCase__										=    None
    UpperCAmelCase__										=    None

   UpperCAmelCase__										=    nn.KLDivLoss(reduction="""batchmean""" )
   UpperCAmelCase__										=    2.0
   UpperCAmelCase__										=    hparams.alpha_mlm
   UpperCAmelCase__										=    hparams.alpha_ce
   UpperCAmelCase__										=    hparams.alpha_hid
   gc.collect()
   torch.cuda.empty_cache()


  def 							SCREAMING_SNAKE_CASE__			(						self    :	Optional[int]						,		_UpperCAmelCase    :	Tuple						,		_UpperCAmelCase    :	Optional[Any]						,		_UpperCAmelCase    :	Dict ):






   """simple docstring"""






   UpperCAmelCase__										=    mask[:, :, None].expand_as(_UpperCAmelCase )
   UpperCAmelCase__										=    s_logits.size(-1 )
   UpperCAmelCase__										=    torch.masked_select(_UpperCAmelCase						,		_UpperCAmelCase )  # (bs * seq_length * voc_size) modulo the 1s in mask
   UpperCAmelCase__										=    torch.masked_select(_UpperCAmelCase						,		_UpperCAmelCase )  # (bs * seq_length * voc_size) modulo the 1s in mask
   UpperCAmelCase__										=    s_logits_slct.view(-1						,		_UpperCAmelCase )  # (bs * seq_length, voc_size) modulo the 1s in mask
   UpperCAmelCase__										=    t_logits_slct.view(-1						,		_UpperCAmelCase )  # (bs * seq_length, voc_size) modulo the 1s in mask
   assert t_logits_slct.size() == s_logits_slct.size()
   UpperCAmelCase__										=    (
       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 							SCREAMING_SNAKE_CASE__			(						_UpperCAmelCase    :	Optional[Any]						,		_UpperCAmelCase    :	Tuple ):






   """simple docstring"""






   SummarizationModule.add_model_specific_args(_UpperCAmelCase						,		_UpperCAmelCase )
   add_distill_args(_UpperCAmelCase )
   return parser


  def 							SCREAMING_SNAKE_CASE__			(						self    :	Optional[Any]						,		_UpperCAmelCase    :	dict ):






   """simple docstring"""






   UpperCAmelCase__										=    self.tokenizer.pad_token_id
   UpperCAmelCase__							,			UpperCAmelCase__							,			UpperCAmelCase__										=    batch["""input_ids"""], batch["""attention_mask"""], batch["""labels"""]
   if isinstance(self.model						,		_UpperCAmelCase ):
    UpperCAmelCase__										=    self.model._shift_right(_UpperCAmelCase )
   else:
    UpperCAmelCase__										=    shift_tokens_right(_UpperCAmelCase						,		_UpperCAmelCase )

   # noinspection PyCallingNonCallable
   UpperCAmelCase__										=    self(
       _UpperCAmelCase						,		attention_mask=_UpperCAmelCase						,		decoder_input_ids=_UpperCAmelCase						,		output_hidden_states=self.do_calc_hidden_loss						,		output_attentions=_UpperCAmelCase						,		use_cache=_UpperCAmelCase						,		)
   UpperCAmelCase__										=    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
    UpperCAmelCase__										=    nn.CrossEntropyLoss(ignore_index=_UpperCAmelCase )
    UpperCAmelCase__										=    loss_fct(lm_logits.view(-1						,		lm_logits.shape[-1] )						,		labels.view(-1 ) )
   else:
    UpperCAmelCase__										=    nn.functional.log_softmax(_UpperCAmelCase						,		dim=-1 )
    UpperCAmelCase__							,			UpperCAmelCase__										=    label_smoothed_nll_loss(
        _UpperCAmelCase						,		_UpperCAmelCase						,		self.hparams.label_smoothing						,		ignore_index=_UpperCAmelCase )

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

   UpperCAmelCase__										=    student_outputs[
       """encoder_last_hidden_state"""
   ]  # use this unless self.different_base_models
   UpperCAmelCase__							,			UpperCAmelCase__										=    zero_tensor(), zero_tensor()
   if self.different_encoder:  # compute encoder hidden state loss
    UpperCAmelCase__										=    self.teacher.get_encoder()(
        _UpperCAmelCase						,		attention_mask=_UpperCAmelCase						,		output_hidden_states=self.do_calc_hidden_loss						,		)
    if self.different_base_models:
     UpperCAmelCase__										=    all_teacher_encoder_outputs["""last_hidden_state"""]
    elif self.do_calc_hidden_loss:
     UpperCAmelCase__										=    self.calc_hidden_loss(
         _UpperCAmelCase						,		student_outputs["""encoder_hidden_states"""]						,		all_teacher_encoder_outputs["""hidden_states"""]						,		self.e_matches						,		normalize_hidden=self.hparams.normalize_hidden						,		)

   UpperCAmelCase__										=    self.teacher(
       _UpperCAmelCase						,		attention_mask=_UpperCAmelCase						,		encoder_outputs=(teacher_enc_outputs,)						,		decoder_input_ids=_UpperCAmelCase						,		output_hidden_states=self.do_calc_hidden_loss						,		use_cache=_UpperCAmelCase						,		)
   UpperCAmelCase__										=    decoder_input_ids.ne(_UpperCAmelCase )
   UpperCAmelCase__										=    self.calc_ce_loss(_UpperCAmelCase						,		_UpperCAmelCase						,		teacher_outputs["""logits"""] )
   if self.do_calc_hidden_loss:  # Intermediate supervision of decoder hidden states
    UpperCAmelCase__										=    self.calc_hidden_loss(
        _UpperCAmelCase						,		student_outputs["""decoder_hidden_states"""]						,		teacher_outputs["""decoder_hidden_states"""]						,		self.d_matches						,		normalize_hidden=self.hparams.normalize_hidden						,		)

   UpperCAmelCase__										=    (
       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 							SCREAMING_SNAKE_CASE__			(						_UpperCAmelCase    :	List[Any]						,		_UpperCAmelCase    :	Optional[int]						,		_UpperCAmelCase    :	Tuple						,		_UpperCAmelCase    :	Optional[int]						,		_UpperCAmelCase    :	List[Any] ):






   """simple docstring"""






   UpperCAmelCase__										=    """expected list or tuple for hidden_states, got tensor of shape: """
   assert not isinstance(_UpperCAmelCase						,		torch.Tensor ), f'''{msg}{hidden_states.shape}'''
   assert not isinstance(_UpperCAmelCase						,		torch.Tensor ), f'''{msg}{hidden_states_T.shape}'''
   UpperCAmelCase__										=    attention_mask.to(hidden_states[0] )
   UpperCAmelCase__										=    mask.sum() * hidden_states[0].size(-1 )
   UpperCAmelCase__										=    torch.stack([hidden_states[i] for i in range(len(_UpperCAmelCase ) )] )
   UpperCAmelCase__										=    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:
    UpperCAmelCase__										=    nn.functional.layer_norm(_UpperCAmelCase						,		student_states.shape[1:] )
    UpperCAmelCase__										=    nn.functional.layer_norm(_UpperCAmelCase						,		teacher_states.shape[1:] )
   UpperCAmelCase__										=    nn.functional.mse_loss(_UpperCAmelCase						,		_UpperCAmelCase						,		reduction="""none""" )
   UpperCAmelCase__										=    (mse * mask.unsqueeze(0 ).unsqueeze(-1 )).sum() / valid_count
   return masked_mse

def    _UpperCamelCase      (				SCREAMING_SNAKE_CASE__			:      List[Any]       ):
 '''simple docstring'''





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

class 						lowerCAmelCase_   ( lowerCamelCase_   ):







  '''simple docstring'''







  lowerCAmelCase_				:					Tuple				     =						"""translation"""
  lowerCAmelCase_				:					Any				     =						["""bleu"""]
  lowerCAmelCase_				:					str				     =						"""bleu"""


  def __init__(						self    :	Dict						,		_UpperCAmelCase    :	Union[str, Any]						,		**_UpperCAmelCase    :	Union[str, Any] ):






   """simple docstring"""






   super().__init__(_UpperCAmelCase						,		**_UpperCAmelCase )
   assert hparams.src_lang is not None
   assert hparams.tgt_lang is not None
   UpperCAmelCase__										=    hparams.src_lang
   UpperCAmelCase__										=    hparams.tgt_lang
   if self.model.config.decoder_start_token_id is None and isinstance(self.tokenizer						,		_UpperCAmelCase ):
    UpperCAmelCase__										=    self.tokenizer.lang_code_to_id[hparams.tgt_lang]


  def 							SCREAMING_SNAKE_CASE__			(						self    :	Optional[int]						,		_UpperCAmelCase    :	Optional[Any]						,		_UpperCAmelCase    :	List[str] ):






   """simple docstring"""






   return calculate_bleu(_UpperCAmelCase						,		_UpperCAmelCase )



  @staticmethod
  def 							SCREAMING_SNAKE_CASE__			(						_UpperCAmelCase    :	Union[str, Any]						,		_UpperCAmelCase    :	Dict ):






   """simple docstring"""






   TranslationModule.add_model_specific_args(_UpperCAmelCase						,		_UpperCAmelCase )
   add_distill_args(_UpperCAmelCase )
   return parser

def    _UpperCamelCase      (				SCREAMING_SNAKE_CASE__			:      Any       ):
 '''simple docstring'''





 if args.no_teacher:
  UpperCAmelCase__										=    TranslationModule if """translation""" in args.task else SummarizationModule
 else:  # DISTILL WITH TEACHER
  UpperCAmelCase__										=    TranslationDistiller if """translation""" in args.task else SummarizationDistiller
 UpperCAmelCase__										=    module_cls.__name__
 print(F'''using module {args.setup_cls}'''       )
 UpperCAmelCase__										=    module_cls(SCREAMING_SNAKE_CASE__       )
 return model

def    _UpperCamelCase      (				SCREAMING_SNAKE_CASE__			:      Any       ):
 '''simple docstring'''





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

 UpperCAmelCase__										=    create_module(SCREAMING_SNAKE_CASE__       )
 return ft_main(SCREAMING_SNAKE_CASE__       ,						model=SCREAMING_SNAKE_CASE__       )


if __name__ == "__main__":
    UpperCAmelCase_       				=      argparse.ArgumentParser()
    UpperCAmelCase_       				=      pl.Trainer.add_argparse_args(parser)
    UpperCAmelCase_       				=      SummarizationDistiller.add_model_specific_args(parser, os.getcwd())
    UpperCAmelCase_       				=      parser.parse_args()

    distill_main(args)






'''simple docstring'''




from __future__ import annotations

import random
import string

class 						lowerCAmelCase_   :







  '''simple docstring'''







  def __init__(						self    :	int						,		_UpperCAmelCase    :	str | None = None ):






   """simple docstring"""






   UpperCAmelCase__										=    passcode or self.__passcode_creator()
   UpperCAmelCase__										=    self.__make_key_list()
   UpperCAmelCase__										=    self.__make_shift_key()


  def __str__(						self    :	int ):






   """simple docstring"""






   return "".join(self.__passcode )


  def 							SCREAMING_SNAKE_CASE__			(						self    :	List[str]						,		_UpperCAmelCase    :	list[int] ):






   """simple docstring"""






   for i in range(1						,		len(_UpperCAmelCase )						,		2 ):
    iterlist[i] *= -1
   return iterlist


  def 							SCREAMING_SNAKE_CASE__			(						self    :	int ):






   """simple docstring"""






   UpperCAmelCase__										=    string.ascii_letters + string.digits
   UpperCAmelCase__										=    [random.choice(_UpperCAmelCase ) for _ in range(random.randint(10						,		20 ) )]
   return password


  def 							SCREAMING_SNAKE_CASE__			(						self    :	Tuple ):






   """simple docstring"""






   UpperCAmelCase__										=    (
       string.ascii_letters + string.digits + string.punctuation + """ \t\n"""
   )

   UpperCAmelCase__										=    []

   # creates points known as breakpoints to break the key_list_options at those
   # points and pivot each substring
   UpperCAmelCase__										=    sorted(set(self.__passcode ) )
   UpperCAmelCase__										=    []

   # algorithm for creating a new shuffled list, keys_l, out of key_list_options
   for i in key_list_options:
    temp_list.extend(_UpperCAmelCase )

    # checking breakpoints at which to pivot temporary sublist and add it into
    # keys_l
    if i in breakpoints or i == key_list_options[-1]:
     keys_l.extend(temp_list[::-1] )
     temp_list.clear()

        # returning a shuffled keys_l to prevent brute force guessing of shift key
   return keys_l


  def 							SCREAMING_SNAKE_CASE__			(						self    :	Tuple ):






   """simple docstring"""






   UpperCAmelCase__										=    sum(self.__neg_pos([ord(_UpperCAmelCase ) for x in self.__passcode] ) )
   return num if num > 0 else len(self.__passcode )


  def 							SCREAMING_SNAKE_CASE__			(						self    :	Optional[Any]						,		_UpperCAmelCase    :	str ):






   """simple docstring"""






   UpperCAmelCase__										=    """"""

   # decoding shift like Caesar cipher algorithm implementing negative shift or
   # reverse shift or left shift
   for i in encoded_message:
    UpperCAmelCase__										=    self.__key_list.index(_UpperCAmelCase )
    decoded_message += self.__key_list[
        (position - self.__shift_key) % -len(self.__key_list )
    ]

   return decoded_message



  def 							SCREAMING_SNAKE_CASE__			(						self    :	List[str]						,		_UpperCAmelCase    :	str ):






   """simple docstring"""






   UpperCAmelCase__										=    """"""

   # encoding shift like Caesar cipher algorithm implementing positive shift or
   # forward shift or right shift
   for i in plaintext:
    UpperCAmelCase__										=    self.__key_list.index(_UpperCAmelCase )
    encoded_message += self.__key_list[
        (position + self.__shift_key) % len(self.__key_list )
    ]

   return encoded_message

def    _UpperCamelCase      (				SCREAMING_SNAKE_CASE__			:      str = "Hello, this is a modified Caesar cipher"       ):
 '''simple docstring'''





 UpperCAmelCase__										=    ShuffledShiftCipher()
 return cipa.decrypt(cipa.encrypt(SCREAMING_SNAKE_CASE__       )       )


if __name__ == "__main__":
    import doctest

    doctest.testmod()
